You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "YAshhh29 (via GitHub)" <gi...@apache.org> on 2026/08/13 20:46:05 UTC
[PR] Do not replay cached AI steps that cannot be verified [airflow]
YAshhh29 opened a new pull request, #71575:
URL: https://github.com/apache/airflow/pull/71575
When durable execution replays a cached model response or tool result, it compares a fingerprint of the current request against the fingerprint stored with the cache entry. Requests that cannot be serialized to JSON fingerprint as `None`, and `None == None` is `True` — so the guard passed for precisely the requests it could not verify, and a stale entry was replayed.
```diff
- if cached_fingerprint == fingerprint:
+ if fingerprint is not None and cached_fingerprint == fingerprint:
```
**This is a behaviour change, not only a bug fix.** A step whose request cannot be fingerprinted now re-runs live on retry rather than replaying, which costs a model call. The existing code was ambiguous about which behaviour was intended: `fingerprint.py` logged *"cached responses for this step replay without verification"* (fail open), while the `reason=` branch immediately below the comparison explains *"entry predates fingerprinting or the request could not be fingerprinted"* — a message that could never be reached, because that case short-circuited into the replay path. This PR resolves the contradiction in favour of not replaying what cannot be verified, and documents the contract in `agent.rst`.
### Behaviour before and after
The screenshots below drive the real `CachingModel` and `CachingToolset` directly (no pytest, no mocked guard). Each run prints the guard line straight out of `caching_model.py`, so the version under test is visible in the output. The scenario is identical in both: a cached answer for Q3 revenue is stored with a `None` fingerprint, then the step is retried with a request about Q4 revenue that also cannot be fingerprinted.
**Before — guard as it is on `main`**
<img width="1506" height="651" alt="image" src="https://github.com/user-attachments/assets/c2418e4f-5b10-46a5-ba59-cc9a54b3b012" />
The cached Q3 answer is returned in response to the Q4 question, with `live calls: 0` and `replayed steps: 1`. The model is never consulted, and nothing verified that the two requests matched.
**After — guard with this PR**
<img width="1485" height="678" alt="image" src="https://github.com/user-attachments/assets/92d7949c-7872-43d8-9aeb-d2179df207a1" />
The unverifiable step runs live: `live calls: 1`, `replayed steps: 0`, and the answer matches the question that was actually asked.
### Testing
`test_unverifiable_current_request_treated_as_miss` and `test_unverifiable_current_call_treated_as_miss` fail without the change and pass with it.
| `providers/common/ai/tests/unit/common/ai/durable` | result |
| --- | --- |
| guard as it is on `main` | `2 failed, 85 passed` |
| guard with this PR | `87 passed` |
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — GitHub Copilot (Claude Opus 5)
Generated-by: GitHub Copilot (Claude Opus 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4048400716
##########
providers/common/ai/docs/operators/agent.rst:
##########
@@ -300,6 +300,17 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+If a model request or tool call cannot be fingerprinted -- it carries a value
+that will not serialize to JSON -- that step is not cached, and on retry it runs
+live rather than replaying an unverified entry. This is rarely confined to a
+single step: the usual causes are a non-serializable value in ``model_settings``,
Review Comment:
Fair — the `_digest` normalization takes care of the dominant tool-call cause, so that part mostly resolves itself, but the paragraph was still wrong to open by covering both paths and then only give model-path causes.
It's now three parts:
a) normalization first, so a `datetime` or `Decimal` argument isn't presented as a hazard at all,
b) then the model path, where `model_settings` and message history make it run-wide,
c) then tool calls on their own, noting they're fingerprinted from name, args and call id alone and report `could not fingerprint tool call`.
The quoted strings now match what each path actually emits.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fingerprint durable agent steps that hold sets or values that are not JSON [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-6202277308
Merged main in, to clear the conflicts with the code mode changes. I kept your new Code mode section and the `run_code` timestamp rule alongside this PR's docs, with no code changes otherwise. The durable tests pass on pydantic-ai 2.36 and 2.48.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4105747768
##########
providers/common/ai/docs/operators/agent.rst:
##########
@@ -300,6 +300,23 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from values normalized through pydantic, so ordinary
+types that are not JSON -- a ``datetime`` or ``Decimal`` tool argument, a
+dataclass in ``tool_choice`` -- still fingerprint normally. If a value cannot be
+serialized even then, that step is not cached, and on retry it runs live rather
+than replaying an unverified entry.
+
+On the model path this is rarely confined to a single step: the causes are such a
+value in ``model_settings``, which is attached to every request, or in the message
+history, which every later request carries forward. Either one degrades all
+subsequent model steps the same way, leaving durable execution with nothing to
+replay, so the retry re-runs the agent at full cost. The
+``could not fingerprint model request`` warning names the step where this began.
+
+A tool call is fingerprinted from its name, arguments and call id alone, so
+neither of those causes reaches it. One that cannot be fingerprinted is reported
+as ``could not fingerprint tool call`` and costs only that call.
Review Comment:
You're right — and it contradicted step 3 two paragraphs up. It now says a tool call that can't be fingerprinted costs that call and, if the live re-run returns something different from the first attempt, the model steps after it too, since the result becomes part of the history they fingerprint. It points back to step 3 rather than describing the cascade twice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-5835805754
Thanks @kaxil, all five are addressed. I rebased onto main, so this is a force-push and the SHAs in my earlier comments are stale. The new commits are `3c4c1b2c` (canonical form), `db2f5ae1` (`error=` in the warnings), `ebab1f46` (docs) and `209a06a0` (a spelling fix).
The iterator and `set[str]` issues were regressions from my `to_jsonable_python` change, not pre-existing bugs, so thanks for catching them. Details for each are in the thread replies.
Beyond what you flagged:
- a set coming back from a tool was also unstable, so messages now dump in `mode="python"`. On a 451-message history this gives the same digest as main, so nothing in flight moves.
- that change opened two ways to escape the handler: `AttributeError` from a malformed message, and `RecursionError` from a circular reference. Both now degrade to an uncached step.
The docs moved to `durable_execution.rst` with #73523, so the paragraphs landed there. CI is green: 118 durable tests (88 on main).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-6088707709
@kaxil thanks, all five are addressed in one commit on top (`cb7afc5c`). Before that I squashed the old history into 3 commits and rebased onto main, keeping your `_normalize_params` from #74314 as is, so the earlier SHAs are stale.
The short version: a list is sorted only where it really holds the set's members, anything python mode can't dump falls back to main's hashing, and non-string keys are checked by rendering them on their own. Details are in each thread. Tested on pydantic-ai 2.33 and 2.44, and on pydantic 2.12.0 and 2.13.4.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r3972084886
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -34,9 +34,8 @@
Fields that pydantic-ai regenerates on every attempt (message-level
``timestamp``/``run_id``/``conversation_id`` and part-level ``timestamp``)
are excluded from the fingerprint. Requests that cannot be serialized to
-JSON fingerprint as ``None``, which degrades that step to unverified
-positional replay (the pre-fingerprint behavior) rather than disabling
-caching.
+JSON fingerprint as ``None`` and re-run live instead of replaying without
Review Comment:
Thanks @kaxil — all three addressed in `98004ea265`.
**1. Skipping the save when `fingerprint is None`** — done, in both `caching_model.py` and `caching_toolset.py`. The entry is unreachable exactly as you describe: the guard needs a non-`None` current fingerprint, and a real digest never equals a stored `None`, so the write only bought a dead entry per step plus a full rewrite of the cache blob each time. The step now returns before the save.
One consequence worth noting: the warning below the comparison is now reachable only for entries that predate fingerprinting, since a `None` fingerprint is no longer something we can write. I also stopped incrementing `cached_model` / `cached_tool` on that path — otherwise the INFO summary reports steps as cached when nothing was persisted.
Happy to drop that part if you'd rather keep this commit strictly to the dead-write fix.
**2. "That step" undersells it** — agreed, and it matches what I see putting those cases through `fingerprint_model_request`. The docs now state that the realistic causes — a non-JSON value in `model_settings`, which accompanies every request, or in the message history, which every later request carries forward — degrade every subsequent model step, leaving durable execution with nothing to replay and a retry that re-runs the agent at full cost. It also points at the `could not fingerprint model request` warning as the marker for where it began.
**3. The `_TRANSPORT_ONLY_SETTINGS` comment** — updated to the current consequence: the frozenset is load-bearing, because one non-JSON member now costs caching and replay for the whole run rather than just the verification. I carried the same correction into the module docstring, `fingerprint_model_request`'s docstring, both `log.warning` strings (they still read "this step will execute live on retry"), and the docstring on `test_httpx_timeout_does_not_disable_fingerprint`, which had the same stale framing.
**Tests** — the two existing unverifiable-step tests now assert that nothing is written, plus one new test per wrapper for the fresh-cache case. All four fail without this change and pass with it.
One adjacent thing I noticed but deliberately left out of scope: `DurableStorage.save_tool_result` skips the write for a non-serializable result, but the caller still counts it as cached, so the same counter skew exists on that path. Want that folded in here, or kept separate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4190202788
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -102,19 +159,201 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _check_guide(guide: Any) -> bool:
+ """
+ Refuse an iterator, and report whether the JSON rendering needs ``_order_sets``.
+
+ ``guide`` is pydantic's python-mode dump of the payload. It has the shape of the
+ json-mode rendering that is hashed -- the same keys, aliases, exclusions,
+ computed fields, extras and serializer output -- but keeps sets as sets and dict
+ keys as they are, and wraps an iterator in a ``SerializationIterator`` without
+ reading it. So this walk runs no user code, and nothing has been consumed yet.
+
+ Raises ``TypeError`` if the payload renders through an iterator anywhere.
+ Rendering it to JSON would consume it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into an iterator; a tool return that holds a generator
+ would reach the model empty. Returns ``True`` if the guide holds a set, or a dict
+ key that is not a string, since only those can make the JSON rendering depend on
+ the hash seed or lose a key.
+ """
+ needs_check = False
+ pending = [guide]
+ seen: set[int] = set()
+ while pending:
+ item = pending.pop()
+ if id(item) in seen:
+ # The dump shares this object, or left a cycle in place: walked already.
+ continue
+ seen.add(id(item))
+ children: Iterable[Any]
+ if isinstance(item, dict):
+ needs_check = needs_check or any(not isinstance(key, str) for key in item)
+ children = item.values()
+ elif isinstance(item, (list, tuple, deque)):
+ children = item
+ elif isinstance(item, (set, frozenset)):
+ needs_check = True
+ children = item
+ elif isinstance(item, Iterator):
+ raise TypeError(f"cannot fingerprint a value that renders through a {type(item).__name__}")
+ elif isinstance(item, enum.Enum):
+ # Python mode keeps an Enum member; JSON renders its value.
+ children = (item.value,)
+ else:
+ continue
+ pending.extend(child for child in children if type(child) not in _LEAF_TYPES)
+ return needs_check
+
+
+def _json_order(member: Any) -> str:
+ return json.dumps(member, sort_keys=True)
+
+
+def _template(value: Any) -> Any:
+ """
+ Say where the sets are inside one set member, or ``None`` if that cannot be said.
+
+ ``_LEAF`` for a member with no set inside it, ``("set", inner)`` for a set whose
+ members all have the template ``inner``, and ``("sequence", parts)`` for a tuple
+ with a set somewhere in it.
+ """
+ if isinstance(value, enum.Enum):
+ return _template(value.value)
+ if isinstance(value, (set, frozenset)):
+ inner = _member_template(value)
+ return None if inner is None else ("set", inner)
+ if isinstance(value, (list, tuple, deque)):
+ parts = tuple(_template(item) for item in value)
+ if None in parts:
+ return None
+ return _LEAF if all(part == _LEAF for part in parts) else ("sequence", parts)
+ if isinstance(value, dict):
+ # No set member dumps to a dict (python mode refuses a set of models), so
+ # nothing is known about one.
+ return None
+ return _LEAF
+
+
+def _member_template(members: Iterable[Any]) -> Any:
+ """Return the template every member of a set shares, or ``None`` if they differ."""
+ templates = {_template(member) for member in members}
+ if not templates:
+ return _LEAF
+ return templates.pop() if len(templates) == 1 else None
+
+
+def _apply_template(template: Any, rendered: Any) -> Any:
+ """Sort the lists that ``template`` puts a set at, in the rendering of one set member."""
+ if template == _LEAF or not isinstance(rendered, list):
+ return rendered
+ kind, inner = template
+ if kind == "set":
+ return sorted((_apply_template(inner, item) for item in rendered), key=_json_order)
+ if len(rendered) != len(inner):
+ return rendered
+ return [_apply_template(part, item) for part, item in zip(inner, rendered)]
+
+
+def _order_sets(guide: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is the json-mode rendering that is hashed, and ``guide`` the
+ python-mode dump of the same payload (see ``_check_guide``). They are walked
+ side by side: a dict or a sequence pairs up by position, because both dumps
+ keep the same order, and a list is sorted where the guide holds a set. A set's
+ members cannot pair up by position -- the dump builds a new set, which may
+ iterate in another order -- so the members' shared template says where any set
+ inside one of them sits (see ``_template``). Anything that does not line up is
+ kept exactly as rendered, so nothing that did not come from a set is reordered.
+
+ Raises ``TypeError`` where distinct dict keys render alike, such as ``1`` and
+ ``"1"``, or ``None`` and ``nan`` in a message dump: the guide then holds more keys
+ than the rendering, and hashing it would let two different payloads share a
+ digest. A dict whose keys are all strings cannot collide, so one that renders
+ fewer keys was reshaped by a serializer that applies only in JSON mode, and is
+ kept as rendered.
+ """
+ if isinstance(guide, enum.Enum):
+ # An Enum member renders as its value.
+ return _order_sets(guide.value, rendered)
+ if isinstance(guide, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(guide):
+ return rendered
+ template = _member_template(guide)
+ if template is None:
+ return rendered
+ return sorted((_apply_template(template, item) for item in rendered), key=_json_order)
+ if isinstance(guide, dict):
+ if not isinstance(rendered, dict):
+ return rendered
+ if len(rendered) != len(guide):
+ if len(rendered) < len(guide) and any(not isinstance(key, str) for key in guide):
+ raise TypeError("dict keys collide once rendered as JSON")
Review Comment:
Two gaps remain for non-string keys in this block. A JSON-only serializer that filters a `dict[int, int]` (dropping zero counts, say) lands here as "dict keys collide" even though nothing collides, and since it's in a tool return, the rest of the run fingerprints `None` where main hashed it. Raising only when rendering the keys on their own, `to_jsonable_python(dict.fromkeys(guide), bytes_mode="base64")`, gives fewer keys than the guide would keep the `{1: "a", "1": "b"}` refusal. Separately, the bail-out at 296 only checks string keys, so a JSON-only serializer that reorders an int-keyed dict pairs a set with its neighbour's list. `{2: {"p", "q"}, 1: ["y", "x"]}` and the same with `["x", "y"]` then share a digest. Also bailing out for a non-string `guide_key` whose rendering `next(iter(to_jsonable_python({guide_key: None})))` differs from `key` closes it. That contradicts the docstring's "nothing that did not come from a set is reordered" at 268.
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -107,6 +107,58 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Plain JSON arguments and settings are fingerprinted exactly as before. Anything
+else is fingerprinted from pydantic's JSON rendering, so ordinary types that are
+not JSON -- a ``datetime`` or ``Decimal`` tool argument, a dataclass in
+``tool_choice``, the bytes in a ``BinaryContent``, a dict keyed by date --
+fingerprint normally. Bytes in tool arguments and settings are rendered as base64,
+so binary data that is not valid UTF-8 fingerprints too; bytes inside a pydantic
+model follow that model's ``ser_json_bytes`` setting instead, which renders them as
+UTF-8 text by default. Because a tool call is fingerprinted from how its arguments
+render, a field excluded from serialization and a secret value, which renders
+masked, take no part in it.
+
+A step whose request cannot be fingerprinted is not cached, and on retry it runs
+live rather than replaying an unverified entry. That happens when pydantic cannot
Review Comment:
Pydantic does render both of these in JSON mode: `{F(x=1), F(x=2)}` becomes a list of dicts, and a model key becomes its str form. The refusal comes from the python-mode dump the fingerprint uses to find sets, and it covers frozen dataclasses too, so "pydantic cannot render" points readers at the wrong thing. If the fallback on `fingerprint.py:376` lands, this only applies to tool arguments, where main returned `None` anyway. If it doesn't, it's a second exception to "hashes the same way as before" at line 134.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -123,44 +362,86 @@ def fingerprint_model_request(
output mode and schema, native tools, ...) so any change to what is sent
to the model invalidates the cached response.
- Returns ``None`` when the request cannot be serialized; ``None`` compares
- equal to ``None``, so requests that cannot be fingerprinted degrade to
- unverified positional replay rather than disabling caching.
+ Returns ``None`` when the request cannot be fingerprinted, which prevents the
+ step from being replayed or cached. Model settings, the tool definitions in the
+ request parameters and the message history are normally carried into every
+ later request, so such a value in any of them usually degrades every
+ subsequent model step of the run the same way. ``step`` is attached to the
+ warning so the log names where it began.
"""
try:
+ # Messages and parameters are hashed from pydantic's json-mode dump, as they
+ # always were, so stored fingerprints still match. The python-mode dumps only
+ # guide ``_order_sets``, which runs where a set or a non-string key needs it.
+ messages_guide = ModelMessagesTypeAdapter.dump_python(messages)
Review Comment:
These python-mode dumps raise where the json-mode dump below succeeds. Python mode dumps a model or dataclass to a dict even when it's a set member or a dict key, and that fails with `TypeError: unhashable type: 'dict'`. So a tool that returns a frozenset of frozen dataclasses or models, or a model with `by_region: dict[Region, float]` where `Region` is frozen, makes this request fingerprint `None`. Since the return stays in the history, every later model step does too, and the rest of the agent re-runs live on retry. Main hashed all of these with a digest that was stable across hash seeds, and so did f3ebf75. The dict-key case also depends on the pydantic version: 2.12.0 (the floor) keeps the key object in python mode and fingerprints it, while 2.13.5 raises, so a test would need to cover both. Wrapping each of the two guide dumps in `except TypeError` and hashing that json-mode dump unordered gave main's digest for every case I tried. `_render` can keep raising, because main ret
urned `None` for those tool args and settings anyway. The comment at 373-375 ("only guide `_order_sets`") would need correcting either way.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -102,19 +159,201 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _check_guide(guide: Any) -> bool:
+ """
+ Refuse an iterator, and report whether the JSON rendering needs ``_order_sets``.
+
+ ``guide`` is pydantic's python-mode dump of the payload. It has the shape of the
+ json-mode rendering that is hashed -- the same keys, aliases, exclusions,
+ computed fields, extras and serializer output -- but keeps sets as sets and dict
+ keys as they are, and wraps an iterator in a ``SerializationIterator`` without
+ reading it. So this walk runs no user code, and nothing has been consumed yet.
+
+ Raises ``TypeError`` if the payload renders through an iterator anywhere.
+ Rendering it to JSON would consume it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into an iterator; a tool return that holds a generator
+ would reach the model empty. Returns ``True`` if the guide holds a set, or a dict
+ key that is not a string, since only those can make the JSON rendering depend on
+ the hash seed or lose a key.
+ """
+ needs_check = False
+ pending = [guide]
+ seen: set[int] = set()
+ while pending:
+ item = pending.pop()
+ if id(item) in seen:
+ # The dump shares this object, or left a cycle in place: walked already.
+ continue
+ seen.add(id(item))
+ children: Iterable[Any]
+ if isinstance(item, dict):
+ needs_check = needs_check or any(not isinstance(key, str) for key in item)
+ children = item.values()
+ elif isinstance(item, (list, tuple, deque)):
+ children = item
+ elif isinstance(item, (set, frozenset)):
+ needs_check = True
+ children = item
+ elif isinstance(item, Iterator):
+ raise TypeError(f"cannot fingerprint a value that renders through a {type(item).__name__}")
+ elif isinstance(item, enum.Enum):
+ # Python mode keeps an Enum member; JSON renders its value.
+ children = (item.value,)
+ else:
+ continue
+ pending.extend(child for child in children if type(child) not in _LEAF_TYPES)
+ return needs_check
+
+
+def _json_order(member: Any) -> str:
+ return json.dumps(member, sort_keys=True)
+
+
+def _template(value: Any) -> Any:
+ """
+ Say where the sets are inside one set member, or ``None`` if that cannot be said.
+
+ ``_LEAF`` for a member with no set inside it, ``("set", inner)`` for a set whose
+ members all have the template ``inner``, and ``("sequence", parts)`` for a tuple
+ with a set somewhere in it.
+ """
+ if isinstance(value, enum.Enum):
+ return _template(value.value)
+ if isinstance(value, (set, frozenset)):
+ inner = _member_template(value)
+ return None if inner is None else ("set", inner)
+ if isinstance(value, (list, tuple, deque)):
+ parts = tuple(_template(item) for item in value)
+ if None in parts:
+ return None
+ return _LEAF if all(part == _LEAF for part in parts) else ("sequence", parts)
+ if isinstance(value, dict):
+ # No set member dumps to a dict (python mode refuses a set of models), so
+ # nothing is known about one.
+ return None
+ return _LEAF
+
+
+def _member_template(members: Iterable[Any]) -> Any:
+ """Return the template every member of a set shares, or ``None`` if they differ."""
+ templates = {_template(member) for member in members}
+ if not templates:
+ return _LEAF
+ return templates.pop() if len(templates) == 1 else None
+
+
+def _apply_template(template: Any, rendered: Any) -> Any:
+ """Sort the lists that ``template`` puts a set at, in the rendering of one set member."""
+ if template == _LEAF or not isinstance(rendered, list):
+ return rendered
+ kind, inner = template
+ if kind == "set":
+ return sorted((_apply_template(inner, item) for item in rendered), key=_json_order)
+ if len(rendered) != len(inner):
+ return rendered
+ return [_apply_template(part, item) for part, item in zip(inner, rendered)]
+
+
+def _order_sets(guide: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is the json-mode rendering that is hashed, and ``guide`` the
+ python-mode dump of the same payload (see ``_check_guide``). They are walked
+ side by side: a dict or a sequence pairs up by position, because both dumps
+ keep the same order, and a list is sorted where the guide holds a set. A set's
+ members cannot pair up by position -- the dump builds a new set, which may
+ iterate in another order -- so the members' shared template says where any set
+ inside one of them sits (see ``_template``). Anything that does not line up is
+ kept exactly as rendered, so nothing that did not come from a set is reordered.
+
+ Raises ``TypeError`` where distinct dict keys render alike, such as ``1`` and
+ ``"1"``, or ``None`` and ``nan`` in a message dump: the guide then holds more keys
+ than the rendering, and hashing it would let two different payloads share a
+ digest. A dict whose keys are all strings cannot collide, so one that renders
+ fewer keys was reshaped by a serializer that applies only in JSON mode, and is
+ kept as rendered.
+ """
+ if isinstance(guide, enum.Enum):
+ # An Enum member renders as its value.
+ return _order_sets(guide.value, rendered)
+ if isinstance(guide, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(guide):
+ return rendered
+ template = _member_template(guide)
+ if template is None:
+ return rendered
+ return sorted((_apply_template(template, item) for item in rendered), key=_json_order)
Review Comment:
Pairing the dumps by shape doesn't prove the list at a set's position came from that set. A `when_used="json"` serializer that keeps keys and lengths but moves values gets past it. Take `tags: set[int]` and `rows: list[int]` with a serializer returning `{"tags": data["rows"], "rows": data["tags"]}`: `rows=[1, 2]` and `rows=[2, 1]` get the same digest, both as a tool argument and in a `ToolReturnPart` on the model path, where main tells them apart. On the tool path a stale replay would also need the same call id, so the real exposure is the history. Sorting only where the guide's own JSON rendering, ordered the same way, matches the ordered result kept those two apart and kept a plain `set[str]` stable across seeds. In the other direction, when pairing gives up (members with different shapes such as `{"a", "b", frozenset({"x", "y"})}`, or a serializer that renames a set field), the set stays in hash-seed order. The step is still cached, and on retry it misses with a "diverged" warn
ing, so refusing it there would give a more accurate log. Docs line 139 also needs a caveat: a JSON-only serializer on a `set` field gets its list re-sorted (`["z-first", "a"]` hashes as `["a", "z-first"]`).
##########
providers/common/ai/tests/unit/common/ai/durable/test_fingerprint.py:
##########
@@ -230,3 +267,1267 @@ def test_arg_order_does_not_matter(self):
assert fingerprint_tool_call("t", {"a": 1, "b": 2}, "id1") == fingerprint_tool_call(
"t", {"b": 2, "a": 1}, "id1"
)
+
+
+class TestPydanticNativeValues:
+ """Values that are not JSON types but hash the same on every attempt must still fingerprint.
+
+ Tool arguments reach ``fingerprint_tool_call`` already coerced by pydantic, and
+ ``tool_choice`` accepts a dataclass while genuinely affecting the response, so it
+ cannot be stripped as transport-only. Since a step that cannot be fingerprinted is
+ no longer cached at all, refusing these values would stop an ordinary typed tool
+ from ever being cached.
+ """
+
+ def test_datetime_tool_argument_fingerprints(self):
+ when = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": when}, "id1") is not None
+
+ def test_decimal_tool_argument_fingerprints(self):
+ assert fingerprint_tool_call("t", {"amount": Decimal("10.5")}, "id1") is not None
+
+ def test_datetime_tool_argument_is_stable_and_distinguishing(self):
+ early = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ late = datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": early}, "id1") == fingerprint_tool_call(
+ "t", {"when": early}, "id1"
+ )
+ assert fingerprint_tool_call("t", {"when": early}, "id1") != fingerprint_tool_call(
+ "t", {"when": late}, "id1"
+ )
+
+ def test_tool_choice_dataclass_fingerprints(self):
+ fp = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["my_tool"])},
+ ModelRequestParameters(),
+ )
+
+ assert fp is not None
+
+ def test_tool_choice_dataclass_still_affects_the_fingerprint(self):
+ one = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["a"])},
+ ModelRequestParameters(),
+ )
+ other = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["b"])},
+ ModelRequestParameters(),
+ )
+
+ assert one is not None
+ assert one != other
+
+ def test_value_pydantic_cannot_serialize_still_returns_none(self):
+ """Normalization must not turn a genuinely unserializable value into a hash."""
+ assert fingerprint_tool_call("t", {"v": object()}, "id1") is None
+
+ def test_plain_payload_digest_is_unchanged_by_normalization(self):
+ """Fingerprints recorded before normalization must still match, so cached entries survive."""
+ payload = {"model": "m", "args": {"b": [1, True, None, "x", 2.5]}, "settings": None}
+ pre_normalization = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+ assert _digest(_render(payload)) == pre_normalization
+
+ def test_dict_keys_render_as_a_json_mode_dump_renders_them(self):
+ by_day = {datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}
+
+ assert _render({"series": by_day}) == {"series": {"2026-01-01": 1.5, "2026-01-02": 2.5}}
+ assert _render({1: "a", "b": 2}) == {"1": "a", "b": 2}
+
+ def test_keys_that_collide_once_rendered_are_refused(self):
+ """``1`` and ``"1"`` render alike, so hashing either payload could replay the other."""
+ assert fingerprint_tool_call("t", {"d": {1: "a", "1": "b"}}, "id1") is None
+
+ def test_keys_that_collide_only_in_the_message_dump_are_refused(self):
+ """The message dump renders ``inf`` and ``nan`` keys alike, where pydantic's defaults do not."""
+ fp = fingerprint_model_request(
+ "m", _with_tool_return({math.inf: 100, math.nan: "unpriced"}), None, ModelRequestParameters()
+ )
+
+ assert fp is None
+
+ def test_bytes_that_are_not_utf8_render_as_base64(self):
+ """Decoding bytes as text would raise on an image and stop the tool from being cached."""
+ assert _render({"image": _PNG}) == {"image": base64.urlsafe_b64encode(_PNG).decode()}
+ assert fingerprint_tool_call("t", {"image": _PNG}, "id1") is not None
+
+ def test_bytes_dict_keys_render_as_base64_with_the_sets_under_them_ordered(self):
+ key = base64.urlsafe_b64encode(_PNG).decode()
+
+ assert _render({"d": {_PNG: _MANY}}) == {"d": {key: sorted(_MANY)}}
+
+
+def _json_mode_reference(model_identifier, messages, model_request_parameters, settings=None):
+ """The fingerprint main computes: pydantic's json-mode dump, hashed as is, with plain JSON settings.
+
+ For anything that is not a set, fingerprints must equal this, or entries stored
+ by an earlier version stop matching and the first retry after an upgrade re-runs
+ the whole agent.
+ """
+ dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
+ stripped = [
+ {
+ **{k: v for k, v in message.items() if k not in ("timestamp", "run_id", "conversation_id")},
+ "parts": [{k: v for k, v in part.items() if k != "timestamp"} for part in message["parts"]],
+ }
+ for message in dumped
+ ]
+ params = TypeAdapter(ModelRequestParameters).dump_python(model_request_parameters, mode="json")
+ payload = {"model": model_identifier, "messages": stripped, "settings": settings, "params": params}
+ return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+
+def _with_tool_return(content):
+ return [
+ ModelRequest(parts=[UserPromptPart(content="go")]),
+ ModelResponse(parts=[ToolCallPart(tool_name="t", args={}, tool_call_id="c1")]),
+ ModelRequest(parts=[ToolReturnPart(tool_name="t", content=content, tool_call_id="c1")]),
+ ]
+
+
+class TestMessageHistoryMatchesJsonModeDump:
+ """The message history must hash exactly as pydantic's json-mode dump renders it.
+
+ Each case here is something a hand-written renderer gets wrong: raw bytes (not
+ valid UTF-8), dict keys that are not strings, ``NaN``, tuples, and fields whose
+ serializer only applies in JSON mode, such as the set-typed request parameters
+ and, on pydantic-ai versions that have it, ``InstructionPart.id``.
+ """
+
+ @pytest.mark.parametrize(
+ "messages",
+ [
+ pytest.param(
+ [
+ ModelRequest(
+ parts=[
+ UserPromptPart(content=["look", BinaryContent(data=_PNG, media_type="image/png")])
+ ]
+ )
+ ],
+ id="binary-content-in-prompt",
+ ),
+ pytest.param(_with_tool_return(_PNG), id="bytes-tool-return"),
+ pytest.param(
+ _with_tool_return({datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}),
+ id="date-keyed-tool-return",
+ ),
+ pytest.param(_with_tool_return({1: "a", "b": 2}), id="mixed-key-tool-return"),
+ pytest.param(_with_tool_return({"x": math.nan}), id="nan-tool-return"),
+ pytest.param(
+ _with_tool_return({"when": datetime.datetime(2026, 1, 1)}), id="datetime-tool-return"
+ ),
+ pytest.param([ModelRequest(parts=(UserPromptPart(content="hi"),))], id="tuple-parts"),
+ pytest.param(_with_tool_return({"pair": ("a", 1)}), id="tuple-tool-return"),
+ ],
+ )
+ def test_fingerprint_equals_the_json_mode_digest(self, messages):
+ fp = fingerprint_model_request("m", messages, None, ModelRequestParameters())
+
+ assert fp is not None
+ assert fp == _json_mode_reference("m", messages, ModelRequestParameters())
+
+ def test_request_parameters_hash_as_their_json_mode_dump(self):
+ """The request parameters dump differently in python and json mode: set-typed fields
+ such as ``revealed_tool_names`` on every supported version, and ``InstructionPart.id``
+ on the versions that have it."""
+ seen = {}
+
+ class Spy(FunctionModel):
+ async def request(self, messages, model_settings, model_request_parameters):
+ seen.setdefault("messages", messages)
+ seen.setdefault("params", model_request_parameters)
+ return await super().request(messages, model_settings, model_request_parameters)
+
+ async def respond(messages, info):
+ return ModelResponse(parts=[TextPart(content="ok")])
+
+ Agent(Spy(respond), instructions="Be terse.").run_sync("hi")
+
+ fp = fingerprint_model_request("m", seen["messages"], None, seen["params"])
+
+ assert fp == _json_mode_reference("m", seen["messages"], seen["params"])
+
+ def test_tuple_parts_still_drop_part_timestamps(self):
+ """Message history passed as objects can hold its parts in a tuple."""
+ t1 = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ t2 = datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc)
+
+ def history(timestamp):
+ return [ModelRequest(parts=(UserPromptPart(content="hi", timestamp=timestamp),))]
+
+ assert fingerprint_model_request(
+ "m", history(t1), None, ModelRequestParameters()
+ ) == fingerprint_model_request("m", history(t2), None, ModelRequestParameters())
+
+ def test_set_in_a_tool_return_hashes_as_its_ordered_members(self):
+ assert fingerprint_model_request(
+ "m", _with_tool_return({"tags": {"b", "c", "a"}}), None, ModelRequestParameters()
+ ) == fingerprint_model_request(
+ "m", _with_tool_return({"tags": ["a", "b", "c"]}), None, ModelRequestParameters()
+ )
+
+
+class TestSetMemberOrdering:
+ """Sets must hash in a fixed order rather than the interpreter's iteration order.
+
+ A ``set[str]`` iterates in an order derived from the process hash seed, and every
+ task attempt runs in a fresh process. Hashing that order would produce a digest
+ the next attempt never reproduces, so the step would re-run live on every retry
+ -- worse than declining to cache it, which at least costs nothing extra.
+ """
+
+ def test_set_hashes_as_its_ordered_members(self):
+ assert _render({"tags": {"beta", "alpha"}}) == {"tags": ["alpha", "beta"]}
+
+ def test_set_matches_the_equivalent_list(self):
+ assert _render({"tags": {"alpha", "beta", "gamma"}}) == _render({"tags": ["alpha", "beta", "gamma"]})
+
+ def test_frozenset_matches_set(self):
+ assert _render({"tags": frozenset({"a", "b"})}) == _render({"tags": {"b", "a"}})
+
+ def test_different_members_still_produce_different_digests(self):
+ assert _render({"tags": {"a", "b"}}) != _render({"tags": {"a", "c"}})
+
+ def test_set_nested_inside_a_list(self):
+ assert _render({"filters": [{"z", "y"}]}) == {"filters": [["y", "z"]]}
+
+ def test_set_inside_a_dataclass_field(self):
+ @dataclasses.dataclass
+ class Filter:
+ tags: set
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_set_inside_a_basemodel_field(self):
+ class Filter(pydantic.BaseModel):
+ tags: set[str]
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_digest_is_stable_across_process_hash_seeds(self):
+ """The real proof: two fresh interpreters must agree, as two attempts would.
+
+ In-process comparisons cannot catch a hash-seed dependency, since one process
+ has one seed. The subprocess loads the module by path so it does not pay for
Review Comment:
Loading by path still imports Airflow: `exec_module` runs the module's `airflow.providers.common.ai.utils` imports, which pull in about 450 modules per interpreter. The four seed tests each start four interpreters and took 106 s of the file's 115 s locally. One snippet that prints all the digests, run under two seeds through a shared helper, would test the same thing for a fraction of the CI time.
Separately, two of the `_order_sets` guards can be removed while the suite still passes, so "reverting any of the 27 safeguards fails the suite" in the description doesn't hold for them. Dropping the `return rendered` on a dict length mismatch (`fingerprint.py:290-293`) lets a `when_used="json"` model serializer that re-adds an excluded field get zipped away: `note="DROP TABLE a"` and `note="DROP TABLE b"` next to a set field then share a digest. Returning any single template instead of `None` for mixed members at line 242 also stays green. A test with that serializer asserting distinct digests, and a direct `_member_template(...) is None` assertion for mixed members (rather than a digest test that depends on the seed), would pin both. The length check in `_apply_template` (252-253) is in the same position.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4105739540
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -137,26 +164,40 @@ def fingerprint_model_request(
except (TypeError, ValueError):
# TypeError from json.dumps, ValueError covers PydanticSerializationError
log.warning(
- "Durable: could not fingerprint model request; cached responses for this "
- "step replay without verification"
+ "Durable: could not fingerprint model request; this step will not be cached and will "
+ "execute live on retry. If the cause is in model settings or message history, every "
+ "later model step of this run is affected too",
+ step=step,
Review Comment:
On both warnings. `PydanticSerializationError` already names the offending type, and the iterator refusal raises a `TypeError` that names the type too, so the log now tells you which value is at fault, not just which step.
Adding it turned up a gap going the other way: `_canonical` walks the payload before `json.dumps` can run its own circular-reference check, so a self-referencing value raised `RecursionError` and slipped past `except (TypeError, ValueError)` entirely. A failed fingerprint should cost a re-run, never the task, so `RecursionError` is caught now, with a test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4234275821
##########
providers/common/ai/tests/unit/common/ai/durable/test_fingerprint.py:
##########
@@ -230,3 +267,1267 @@ def test_arg_order_does_not_matter(self):
assert fingerprint_tool_call("t", {"a": 1, "b": 2}, "id1") == fingerprint_tool_call(
"t", {"b": 2, "a": 1}, "id1"
)
+
+
+class TestPydanticNativeValues:
+ """Values that are not JSON types but hash the same on every attempt must still fingerprint.
+
+ Tool arguments reach ``fingerprint_tool_call`` already coerced by pydantic, and
+ ``tool_choice`` accepts a dataclass while genuinely affecting the response, so it
+ cannot be stripped as transport-only. Since a step that cannot be fingerprinted is
+ no longer cached at all, refusing these values would stop an ordinary typed tool
+ from ever being cached.
+ """
+
+ def test_datetime_tool_argument_fingerprints(self):
+ when = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": when}, "id1") is not None
+
+ def test_decimal_tool_argument_fingerprints(self):
+ assert fingerprint_tool_call("t", {"amount": Decimal("10.5")}, "id1") is not None
+
+ def test_datetime_tool_argument_is_stable_and_distinguishing(self):
+ early = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ late = datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": early}, "id1") == fingerprint_tool_call(
+ "t", {"when": early}, "id1"
+ )
+ assert fingerprint_tool_call("t", {"when": early}, "id1") != fingerprint_tool_call(
+ "t", {"when": late}, "id1"
+ )
+
+ def test_tool_choice_dataclass_fingerprints(self):
+ fp = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["my_tool"])},
+ ModelRequestParameters(),
+ )
+
+ assert fp is not None
+
+ def test_tool_choice_dataclass_still_affects_the_fingerprint(self):
+ one = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["a"])},
+ ModelRequestParameters(),
+ )
+ other = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["b"])},
+ ModelRequestParameters(),
+ )
+
+ assert one is not None
+ assert one != other
+
+ def test_value_pydantic_cannot_serialize_still_returns_none(self):
+ """Normalization must not turn a genuinely unserializable value into a hash."""
+ assert fingerprint_tool_call("t", {"v": object()}, "id1") is None
+
+ def test_plain_payload_digest_is_unchanged_by_normalization(self):
+ """Fingerprints recorded before normalization must still match, so cached entries survive."""
+ payload = {"model": "m", "args": {"b": [1, True, None, "x", 2.5]}, "settings": None}
+ pre_normalization = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+ assert _digest(_render(payload)) == pre_normalization
+
+ def test_dict_keys_render_as_a_json_mode_dump_renders_them(self):
+ by_day = {datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}
+
+ assert _render({"series": by_day}) == {"series": {"2026-01-01": 1.5, "2026-01-02": 2.5}}
+ assert _render({1: "a", "b": 2}) == {"1": "a", "b": 2}
+
+ def test_keys_that_collide_once_rendered_are_refused(self):
+ """``1`` and ``"1"`` render alike, so hashing either payload could replay the other."""
+ assert fingerprint_tool_call("t", {"d": {1: "a", "1": "b"}}, "id1") is None
+
+ def test_keys_that_collide_only_in_the_message_dump_are_refused(self):
+ """The message dump renders ``inf`` and ``nan`` keys alike, where pydantic's defaults do not."""
+ fp = fingerprint_model_request(
+ "m", _with_tool_return({math.inf: 100, math.nan: "unpriced"}), None, ModelRequestParameters()
+ )
+
+ assert fp is None
+
+ def test_bytes_that_are_not_utf8_render_as_base64(self):
+ """Decoding bytes as text would raise on an image and stop the tool from being cached."""
+ assert _render({"image": _PNG}) == {"image": base64.urlsafe_b64encode(_PNG).decode()}
+ assert fingerprint_tool_call("t", {"image": _PNG}, "id1") is not None
+
+ def test_bytes_dict_keys_render_as_base64_with_the_sets_under_them_ordered(self):
+ key = base64.urlsafe_b64encode(_PNG).decode()
+
+ assert _render({"d": {_PNG: _MANY}}) == {"d": {key: sorted(_MANY)}}
+
+
+def _json_mode_reference(model_identifier, messages, model_request_parameters, settings=None):
+ """The fingerprint main computes: pydantic's json-mode dump, hashed as is, with plain JSON settings.
+
+ For anything that is not a set, fingerprints must equal this, or entries stored
+ by an earlier version stop matching and the first retry after an upgrade re-runs
+ the whole agent.
+ """
+ dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
+ stripped = [
+ {
+ **{k: v for k, v in message.items() if k not in ("timestamp", "run_id", "conversation_id")},
+ "parts": [{k: v for k, v in part.items() if k != "timestamp"} for part in message["parts"]],
+ }
+ for message in dumped
+ ]
+ params = TypeAdapter(ModelRequestParameters).dump_python(model_request_parameters, mode="json")
+ payload = {"model": model_identifier, "messages": stripped, "settings": settings, "params": params}
+ return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+
+def _with_tool_return(content):
+ return [
+ ModelRequest(parts=[UserPromptPart(content="go")]),
+ ModelResponse(parts=[ToolCallPart(tool_name="t", args={}, tool_call_id="c1")]),
+ ModelRequest(parts=[ToolReturnPart(tool_name="t", content=content, tool_call_id="c1")]),
+ ]
+
+
+class TestMessageHistoryMatchesJsonModeDump:
+ """The message history must hash exactly as pydantic's json-mode dump renders it.
+
+ Each case here is something a hand-written renderer gets wrong: raw bytes (not
+ valid UTF-8), dict keys that are not strings, ``NaN``, tuples, and fields whose
+ serializer only applies in JSON mode, such as the set-typed request parameters
+ and, on pydantic-ai versions that have it, ``InstructionPart.id``.
+ """
+
+ @pytest.mark.parametrize(
+ "messages",
+ [
+ pytest.param(
+ [
+ ModelRequest(
+ parts=[
+ UserPromptPart(content=["look", BinaryContent(data=_PNG, media_type="image/png")])
+ ]
+ )
+ ],
+ id="binary-content-in-prompt",
+ ),
+ pytest.param(_with_tool_return(_PNG), id="bytes-tool-return"),
+ pytest.param(
+ _with_tool_return({datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}),
+ id="date-keyed-tool-return",
+ ),
+ pytest.param(_with_tool_return({1: "a", "b": 2}), id="mixed-key-tool-return"),
+ pytest.param(_with_tool_return({"x": math.nan}), id="nan-tool-return"),
+ pytest.param(
+ _with_tool_return({"when": datetime.datetime(2026, 1, 1)}), id="datetime-tool-return"
+ ),
+ pytest.param([ModelRequest(parts=(UserPromptPart(content="hi"),))], id="tuple-parts"),
+ pytest.param(_with_tool_return({"pair": ("a", 1)}), id="tuple-tool-return"),
+ ],
+ )
+ def test_fingerprint_equals_the_json_mode_digest(self, messages):
+ fp = fingerprint_model_request("m", messages, None, ModelRequestParameters())
+
+ assert fp is not None
+ assert fp == _json_mode_reference("m", messages, ModelRequestParameters())
+
+ def test_request_parameters_hash_as_their_json_mode_dump(self):
+ """The request parameters dump differently in python and json mode: set-typed fields
+ such as ``revealed_tool_names`` on every supported version, and ``InstructionPart.id``
+ on the versions that have it."""
+ seen = {}
+
+ class Spy(FunctionModel):
+ async def request(self, messages, model_settings, model_request_parameters):
+ seen.setdefault("messages", messages)
+ seen.setdefault("params", model_request_parameters)
+ return await super().request(messages, model_settings, model_request_parameters)
+
+ async def respond(messages, info):
+ return ModelResponse(parts=[TextPart(content="ok")])
+
+ Agent(Spy(respond), instructions="Be terse.").run_sync("hi")
+
+ fp = fingerprint_model_request("m", seen["messages"], None, seen["params"])
+
+ assert fp == _json_mode_reference("m", seen["messages"], seen["params"])
+
+ def test_tuple_parts_still_drop_part_timestamps(self):
+ """Message history passed as objects can hold its parts in a tuple."""
+ t1 = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ t2 = datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc)
+
+ def history(timestamp):
+ return [ModelRequest(parts=(UserPromptPart(content="hi", timestamp=timestamp),))]
+
+ assert fingerprint_model_request(
+ "m", history(t1), None, ModelRequestParameters()
+ ) == fingerprint_model_request("m", history(t2), None, ModelRequestParameters())
+
+ def test_set_in_a_tool_return_hashes_as_its_ordered_members(self):
+ assert fingerprint_model_request(
+ "m", _with_tool_return({"tags": {"b", "c", "a"}}), None, ModelRequestParameters()
+ ) == fingerprint_model_request(
+ "m", _with_tool_return({"tags": ["a", "b", "c"]}), None, ModelRequestParameters()
+ )
+
+
+class TestSetMemberOrdering:
+ """Sets must hash in a fixed order rather than the interpreter's iteration order.
+
+ A ``set[str]`` iterates in an order derived from the process hash seed, and every
+ task attempt runs in a fresh process. Hashing that order would produce a digest
+ the next attempt never reproduces, so the step would re-run live on every retry
+ -- worse than declining to cache it, which at least costs nothing extra.
+ """
+
+ def test_set_hashes_as_its_ordered_members(self):
+ assert _render({"tags": {"beta", "alpha"}}) == {"tags": ["alpha", "beta"]}
+
+ def test_set_matches_the_equivalent_list(self):
+ assert _render({"tags": {"alpha", "beta", "gamma"}}) == _render({"tags": ["alpha", "beta", "gamma"]})
+
+ def test_frozenset_matches_set(self):
+ assert _render({"tags": frozenset({"a", "b"})}) == _render({"tags": {"b", "a"}})
+
+ def test_different_members_still_produce_different_digests(self):
+ assert _render({"tags": {"a", "b"}}) != _render({"tags": {"a", "c"}})
+
+ def test_set_nested_inside_a_list(self):
+ assert _render({"filters": [{"z", "y"}]}) == {"filters": [["y", "z"]]}
+
+ def test_set_inside_a_dataclass_field(self):
+ @dataclasses.dataclass
+ class Filter:
+ tags: set
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_set_inside_a_basemodel_field(self):
+ class Filter(pydantic.BaseModel):
+ tags: set[str]
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_digest_is_stable_across_process_hash_seeds(self):
+ """The real proof: two fresh interpreters must agree, as two attempts would.
+
+ In-process comparisons cannot catch a hash-seed dependency, since one process
+ has one seed. The subprocess loads the module by path so it does not pay for
Review Comment:
Fair, the docstring claimed something it didn't do. The four tests are now one snippet that prints every digest, run under two seeds through a cached helper, with a check that the two seeds really iterate a set differently. The whole file runs in about 5 s locally now.
The guards you named are pinned: a JSON-only serializer that adds an excluded field back must give distinct digests, `_member_template(...) is None` is asserted directly for mixed members, and `_apply_template` has a direct test for the length check. I reverted every guard in the ordering code one at a time again and each one fails a test now. I've taken the "27 safeguards" line out of the description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4112446347
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -119,13 +195,21 @@ def fingerprint_model_request(
output mode and schema, native tools, ...) so any change to what is sent
to the model invalidates the cached response.
- Returns ``None`` when the request cannot be serialized; ``None`` compares
- equal to ``None``, so requests that cannot be fingerprinted degrade to
- unverified positional replay rather than disabling caching.
+ Returns ``None`` when the request cannot be serialized even through pydantic,
+ which prevents the step from being replayed or cached. Because model settings
+ and message history are carried into every later request, such a value in
+ either usually degrades every subsequent model step of the run the same way.
+ ``step`` is attached to that warning so the log names where it began.
"""
try:
- dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
- params = _MODEL_REQUEST_PARAMETERS_ADAPTER.dump_python(model_request_parameters, mode="json")
+ # ``mode="python"``, not ``mode="json"``: a json-mode dump renders a set as
+ # a list in iteration order, and a tool that returned a set puts one in the
+ # message history, where it would reach the hash already unstably ordered.
+ # Python mode leaves it a set for ``_canonical`` to order. For values that
+ # are not sets the two modes produce the same digest, so stored
+ # fingerprints are unaffected.
+ dumped = ModelMessagesTypeAdapter.dump_python(messages)
Review Comment:
You're right, and your larger point turned out to be the real fix. I dropped the hand-written renderer: messages and request parameters go back to pydantic's json-mode dump, exactly as on main, and tool args and settings go through `to_jsonable_python(bytes_mode="base64")`. The only things added on top are sorting lists that came from sets and refusing iterators. A PNG in the prompt and a bytes tool return now digest the same as main.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,11 +127,56 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _canonical(value: Any) -> Any:
+ """
+ Render ``value`` as JSON-safe data whose encoding is identical on every attempt.
+
+ ``to_jsonable_python`` on its own is not a safe fingerprint input, for two
+ reasons that both matter here because tool arguments arrive as live Python
+ objects that pydantic has already validated.
+
+ It renders a ``set`` in iteration order, and for string members that order
+ follows the interpreter's hash seed. Every task attempt is a fresh process, so
+ a ``set[str]`` argument would hash differently each time and never replay --
+ worse than declining to cache, because the step re-runs live on every retry.
+ Members are therefore ordered by their own JSON encoding.
+
+ It also *consumes* an iterator, and pydantic validates an ``Iterable[T]``
+ parameter lazily into a ``ValidatorIterator``. Normalizing the arguments would
+ drain the tool's own input before the tool ran, so the tool would see an empty
+ sequence and that wrong result would be cached under the fingerprint of the
+ full one. Such a value is refused instead, which degrades the step to the
+ not-cached path rather than corrupting it.
+ """
+ if value is None or isinstance(value, (str, bool, int, float)):
+ return value
+ if isinstance(value, Iterator):
+ # Hashing this means draining it, leaving nothing for the tool to read.
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ return {key: _canonical(item) for key, item in value.items()}
Review Comment:
Covered by the same change, since keys are rendered by pydantic now, so `dict[date, float]`, enum and UUID keys digest as on main. One case main hashed ambiguously, `{1: "a", "1": "b"}`, now refuses to fingerprint, because both keys render as `"1"`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-6134148275
Rebased onto main to clear the conflict with #74423. It was the module docstring only, and I kept your wording there for which capability ids are left out. No code changes since the last push.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-5607315486
Thanks @kaxil — all three addressed in `98004ea265`.
**1. Skipping the save when `fingerprint is None`** — done, in both `caching_model.py` and `caching_toolset.py`. The entry is unreachable exactly as you describe: the guard needs a non-`None` current fingerprint, and a real digest never equals a stored `None`, so the write only bought a dead entry per step plus a full rewrite of the cache blob each time. The step now returns before the save.
One consequence worth noting: the warning below the comparison is now reachable only for entries that predate fingerprinting, since a `None` fingerprint is no longer something we can write.I also stopped incrementing `cached_model` / `cached_tool` on that path — otherwise the INFO summary reports steps as cached when nothing was persisted.
Happy to drop that part if you'd rather keep this commit strictly to the dead-write fix.
**2. "That step" undersells it** — agreed, and it matches what I see putting those cases through `fingerprint_model_request`. The docs now state that the realistic causes — a non-JSON value in `model_settings`, which accompanies every request, or in the message history, which every later request carries forward — degrade every subsequent model step, leaving durable execution with nothing to replay and a retry that re-runs the agent at full cost. It also points at the `could not fingerprint model request` warning as the marker for where it began.
**3. The `_TRANSPORT_ONLY_SETTINGS` comment** — updated to the current consequence: the frozenset is load-bearing, because one non-JSON member now costs caching and replay for the whole run rather than just the verification. I carried the same correction into the module docstring, `fingerprint_model_request`'s docstring, both `log.warning` strings (they still read "this step will execute live on retry"), and the docstring on `test_httpx_timeout_does_not_disable_fingerprint`, which had the same stale framing.
**Tests** — the two existing unverifiable-step tests now assert that nothing is written, plus one new test per wrapper for the fresh-cache case. All four fail without this change and pass with it.
One adjacent thing I noticed but deliberately left out of scope: `DurableStorage.save_tool_result` skips the write for a non-serializable result, but the caller still counts it as cached, so the same counter skew exists on that path. Want that folded in here, or kept separate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4112451514
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -85,9 +107,16 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
timestamps, part-level timestamps); user data such as tool arguments is
never recursed into, so an argument legitimately named ``run_id`` still
affects the fingerprint.
+
+ Raises ``TypeError`` if a message did not dump to a mapping. A python-mode dump
+ passes an object it does not recognize straight through, and a fingerprint that
+ cannot strip the volatile fields would change on every attempt, so the caller
+ degrades to an unverifiable ``None`` instead.
"""
stripped = []
for message in messages_dump:
+ if not isinstance(message, Mapping):
+ raise TypeError(f"expected a dumped message mapping, got {type(message).__name__}")
cleaned = {k: v for k, v in message.items() if k not in _VOLATILE_MESSAGE_KEYS}
if isinstance(cleaned.get("parts"), list):
Review Comment:
The json-mode dump turns the tuple back into a list, so the timestamp strip applies again. There's a test with tuple parts and two different part timestamps.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -85,9 +107,16 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
timestamps, part-level timestamps); user data such as tool arguments is
never recursed into, so an argument legitimately named ``run_id`` still
affects the fingerprint.
+
+ Raises ``TypeError`` if a message did not dump to a mapping. A python-mode dump
+ passes an object it does not recognize straight through, and a fingerprint that
Review Comment:
Dropped the guard and the test. With the json-mode dump every message comes back as a mapping, and as you said, that input couldn't occur anyway.
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -102,6 +102,30 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from each value's canonical form, so ordinary types that
Review Comment:
Updated. The paragraph now says fingerprints use pydantic's JSON rendering and that older entries still match, the run-wide causes include the tool definitions the request carries, and the `BinaryContent` note mentions the cascade when the re-run returns something different.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4128198297
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,19 +123,107 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _is_dataclass_instance(value: Any) -> bool:
+ return dataclasses.is_dataclass(value) and not isinstance(value, type)
+
+
+def _refuse_iterators(value: Any) -> None:
+ """
+ Raise ``TypeError`` if ``value`` holds an iterator anywhere inside it.
+
+ Rendering an iterator consumes it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into a ``ValidatorIterator``. Tool arguments are
+ fingerprinted before the tool runs, so rendering them would drain the tool's
+ own input: the tool would see an empty sequence, and that wrong result would be
+ cached under the fingerprint of the full one. Refusing degrades the step to the
+ not-cached path instead.
+ """
+ if value is None or isinstance(value, (str, bytes, bytearray, bool, int, float)):
+ return
+ if isinstance(value, Iterator):
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ children: Iterable[Any] = value.values()
+ elif isinstance(value, (list, tuple, set, frozenset)):
+ children = value
+ elif isinstance(value, BaseModel):
+ children = [getattr(value, name, None) for name in type(value).model_fields]
+ elif _is_dataclass_instance(value):
+ children = [getattr(value, field.name, None) for field in dataclasses.fields(value)]
+ else:
+ return
+ for child in children:
+ _refuse_iterators(child)
+
+
+def _order_sets(value: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is pydantic's JSON rendering of ``value`` and is otherwise kept as
+ is, so pydantic stays the only renderer: bytes, dates, dict keys, ``NaN`` and
+ custom serializers come out exactly as a json-mode dump renders them. The one
+ thing pydantic cannot do stably is order a set. It lists the members in
+ iteration order, which for strings follows the interpreter's hash seed, and
+ every task attempt is a fresh process, so a ``set[str]`` would hash differently
+ on each attempt and never replay. ``value`` is walked alongside ``rendered``
+ only to find those lists; a branch whose rendering does not line up with the
+ object, such as one with a custom serializer, is left exactly as rendered.
+ """
+ if isinstance(value, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(value):
+ return rendered
+ members = [_order_sets(member, item) for member, item in zip(value, rendered)]
+ return sorted(members, key=lambda member: json.dumps(member, sort_keys=True))
+ if isinstance(value, Mapping):
+ if not isinstance(rendered, dict):
+ return rendered
+ if len(rendered) != len(value):
+ # Distinct keys that render alike, such as 1 and "1": the digest could no
+ # longer tell those payloads apart, so refuse rather than hash either one.
+ raise TypeError("dict keys collide once rendered as JSON")
Review Comment:
The same positional pairing can also replay a stale step, which is worse than the spurious `None` above. A serializer that reorders keys without dropping any keeps the lengths equal, so the zip at 185-188 pairs each value with a neighbour's rendering. With `@field_serializer("filters")` returning `dict(sorted(v.items()))` on `Query(filters={"status": {3, 7}, "order": ["created", "id"]})`, the set is paired with the rendered `order` list and sorts it, so `order=["created", "id"]` and `order=["id", "created"]` get the same non-`None` `fingerprint_tool_call` digest on every hash seed I tried, where main returns `None` for both. So a length check alone won't cover it. Comparing `list(to_jsonable_python(dict.fromkeys(value), bytes_mode="base64"))` against the rendered keys, raising when it is shorter than `value` and returning `rendered` when the keys differ, gave distinct digests for the two orders, still refused `{1: "a", "1": "b"}`, still sorted a plain set, and rendered the dropped
-`authorization` case unchanged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-5921205985
Thanks @kaxil, all of these are addressed. The extras, aliases and reordered-keys comments came from the same root problem, so I changed the approach instead of patching each case. The fingerprint now pairs pydantic's JSON output with pydantic's own python-mode dump of the same value. That dump has the same keys, aliases, extras and serializer output, but keeps sets as sets and doesn't read iterators, so set ordering, key collisions and the iterator check all come from pydantic itself. Tool args are also rendered from a deep copy now, so fingerprinting can't change what the tool gets, and anything that raises while rendering just costs a live re-run.
I also rebased onto main, so the older SHAs are stale. Since #73873 changed `cached_*` to mean written, a step that can't be fingerprinted now counts as skipped and shows up in the end-of-run warning, and I dropped my "count executed steps" commit because main covers it.
Tested on pydantic-ai 2.33 and 2.44: digests still match main for everything that isn't a set, and reverting any of the 27 safeguards one at a time fails the suite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4150122713
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,19 +123,107 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _is_dataclass_instance(value: Any) -> bool:
+ return dataclasses.is_dataclass(value) and not isinstance(value, type)
+
+
+def _refuse_iterators(value: Any) -> None:
+ """
+ Raise ``TypeError`` if ``value`` holds an iterator anywhere inside it.
+
+ Rendering an iterator consumes it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into a ``ValidatorIterator``. Tool arguments are
+ fingerprinted before the tool runs, so rendering them would drain the tool's
+ own input: the tool would see an empty sequence, and that wrong result would be
+ cached under the fingerprint of the full one. Refusing degrades the step to the
+ not-cached path instead.
+ """
+ if value is None or isinstance(value, (str, bytes, bytearray, bool, int, float)):
+ return
+ if isinstance(value, Iterator):
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ children: Iterable[Any] = value.values()
+ elif isinstance(value, (list, tuple, set, frozenset)):
+ children = value
+ elif isinstance(value, BaseModel):
+ children = [getattr(value, name, None) for name in type(value).model_fields]
+ elif _is_dataclass_instance(value):
+ children = [getattr(value, field.name, None) for field in dataclasses.fields(value)]
+ else:
+ return
+ for child in children:
+ _refuse_iterators(child)
+
+
+def _order_sets(value: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is pydantic's JSON rendering of ``value`` and is otherwise kept as
+ is, so pydantic stays the only renderer: bytes, dates, dict keys, ``NaN`` and
+ custom serializers come out exactly as a json-mode dump renders them. The one
+ thing pydantic cannot do stably is order a set. It lists the members in
+ iteration order, which for strings follows the interpreter's hash seed, and
+ every task attempt is a fresh process, so a ``set[str]`` would hash differently
+ on each attempt and never replay. ``value`` is walked alongside ``rendered``
+ only to find those lists; a branch whose rendering does not line up with the
+ object, such as one with a custom serializer, is left exactly as rendered.
+ """
+ if isinstance(value, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(value):
+ return rendered
+ members = [_order_sets(member, item) for member, item in zip(value, rendered)]
+ return sorted(members, key=lambda member: json.dumps(member, sort_keys=True))
+ if isinstance(value, Mapping):
+ if not isinstance(rendered, dict):
+ return rendered
+ if len(rendered) != len(value):
+ # Distinct keys that render alike, such as 1 and "1": the digest could no
+ # longer tell those payloads apart, so refuse rather than hash either one.
+ raise TypeError("dict keys collide once rendered as JSON")
Review Comment:
You're right on both. It now only refuses when a key isn't a string and the rendering ends up with fewer keys, since that's the only way keys can actually collide. A serializer that drops or reorders keys does the same thing in both dumps, so nothing gets paired with a neighbour's value anymore. The dropped-`authorization` case hashes like main again, your two `order` lists give different digests, and `{1: "a", "1": "b"}` is still refused. Tests cover both, including a serializer nested in the annotation.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -33,24 +33,43 @@
Fields that pydantic-ai regenerates on every attempt (message-level
``timestamp``/``run_id``/``conversation_id`` and part-level ``timestamp``)
-are excluded from the fingerprint. Requests that cannot be serialized to
-JSON fingerprint as ``None``, which degrades that step to unverified
-positional replay (the pre-fingerprint behavior) rather than disabling
-caching.
+are excluded from the fingerprint. Every payload is rendered by pydantic in
+JSON mode before hashing, so values that are not JSON types but render the same
+way on every attempt -- a ``datetime`` or ``Decimal`` tool argument, a dataclass
+in ``tool_choice``, bytes in a ``BinaryContent`` -- still produce a usable
+fingerprint, and the message history hashes exactly as it always has. The one
+correction applied on top is that set members are ordered by their JSON encoding,
+because a set of strings iterates in an order that follows the interpreter's hash
+seed and every task attempt is a fresh process (see ``_order_sets``).
+
+A request that cannot be rendered fingerprints as ``None``: that step is
+neither replayed nor cached, and re-runs live instead of replaying without
+verification. A lazily validated ``Iterable`` argument lands there deliberately,
+since hashing it would consume the input the tool has not read yet, as does a
+value pydantic cannot serialize at all. On the model path this is seldom
+confined to one step, because model settings, the tool definitions and the
+message history are carried into every later request, so durable execution stops
+contributing anything for the rest of the run. A tool call is fingerprinted from its name, arguments and
+call id alone, so it can only lose its own step.
Review Comment:
Fixed. It now says a tool call can't stop another step from being fingerprinted, but a re-run that returns something different still invalidates the model steps after it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r3963555700
##########
providers/common/ai/src/airflow/providers/common/ai/durable/caching_model.py:
##########
@@ -95,7 +95,7 @@ async def request(
cached, cached_fingerprint = self.storage.load_model_response(key)
if cached is not None:
- if cached_fingerprint == fingerprint:
+ if fingerprint is not None and cached_fingerprint == fingerprint:
Review Comment:
An entry stored with `fingerprint=None` can never be replayed now: this line needs `fingerprint is not None`, and a real digest won't equal a stored `None` either. Line 115 still saves with `fingerprint=fingerprint` (same in `caching_toolset.py:92`), so a run that can't fingerprint writes a dead entry per step and reads it back next attempt only to log the warning below, and `ObjectStorageDurableCache._save_cache` rewrites the whole blob each time. Worth skipping the save when `fingerprint is None`?
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -34,9 +34,8 @@
Fields that pydantic-ai regenerates on every attempt (message-level
``timestamp``/``run_id``/``conversation_id`` and part-level ``timestamp``)
are excluded from the fingerprint. Requests that cannot be serialized to
-JSON fingerprint as ``None``, which degrades that step to unverified
-positional replay (the pre-fingerprint behavior) rather than disabling
-caching.
+JSON fingerprint as ``None`` and re-run live instead of replaying without
Review Comment:
The comment on `_TRANSPORT_ONLY_SETTINGS` (line 66) still says an unserializable setting would "silently disable replay verification for every step", which was the old consequence. After this change it disables replay entirely for the run, which makes that frozenset load-bearing: `ModelSettings.timeout` being `float | httpx.Timeout` was already one member that needed stripping, and any other non-JSON member now costs the whole feature rather than just verification. Worth updating that comment along with these docstrings.
##########
providers/common/ai/docs/operators/agent.rst:
##########
@@ -300,6 +300,9 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+If a model request or tool call cannot be fingerprinted, that step runs live on
Review Comment:
"That step" undersells it. The realistic causes are a non-JSON value in model settings (present on every request) or in the message history (which stays there), so it's every model step from that point on, not one, and durable execution ends up doing nothing for the run. I ran both through `fingerprint_model_request` and every subsequent step came back `None`. Worth saying that here so a full-price retry isn't a surprise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4087586622
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -137,26 +164,40 @@ def fingerprint_model_request(
except (TypeError, ValueError):
# TypeError from json.dumps, ValueError covers PydanticSerializationError
log.warning(
- "Durable: could not fingerprint model request; cached responses for this "
- "step replay without verification"
+ "Durable: could not fingerprint model request; this step will not be cached and will "
+ "execute live on retry. If the cause is in model settings or message history, every "
+ "later model step of this run is affected too",
+ step=step,
Review Comment:
Now that `step` is here, the one thing still missing from this warning is what could not be serialized. `PydanticSerializationError` carries the offending type in its message ("Unable to serialize unknown type: <class 'httpx.Timeout'>"), so `except (TypeError, ValueError) as exc` plus `error=str(exc)` here and on the tool warning below would let someone find the setting or argument without a debugger.
##########
providers/common/ai/docs/operators/agent.rst:
##########
@@ -300,6 +300,23 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from values normalized through pydantic, so ordinary
Review Comment:
Two things on this hunk. First, it now conflicts with main: #73523 moved this whole section to `providers/common/ai/docs/durable_execution.rst` (the anchor line "After successful completion, the cached steps are deleted." is at line 103 there, and `agent.rst` now just says "Moved to :doc:`../durable_execution`"), so these three paragraphs need to land there on rebase. Second, "still fingerprint normally" is broader than what the code delivers: a `set` argument fingerprints, but not stably across attempts (see the `_digest` comment). If the canonicalization lands this sentence becomes true as written; otherwise it needs the module docstring's "serialize to the same bytes on every attempt" hedge with `set` named as the exception.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -99,10 +114,18 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
def _digest(payload: Any) -> str:
Review Comment:
One thing the normalization changes on the tool path that the old `json.dumps` did not: pydantic validates an `Iterable[int]` (or any `Iterable[T]`) tool parameter lazily and hands the tool a `ValidatorIterator`, and `to_jsonable_python` drains it to build the list. `call_tool` fingerprints `tool_args` on line 73 and passes the same dict to the wrapped tool on line 93, so the tool sees an exhausted iterator. I ran it in breeze with the installed pydantic-ai and `def total(values: Iterable[int]) -> int: return sum(values)`: the validated arg is a `ValidatorIterator`, after `to_jsonable_python` it yields nothing and the tool returns 0, and that 0 gets cached under the fingerprint of `[1, 2, 3]`. On the merge-base `json.dumps` raised `TypeError` without advancing it, so the tool still got 6. Worth either rejecting iterator-valued args before normalizing (return `None`, which is the not-cached path), or normalizing a deep copy so the original objects reach the tool untouched, plus a r
egression test with a real `FunctionToolset` and an `Iterable[int]` parameter.
##########
providers/common/ai/docs/operators/agent.rst:
##########
@@ -300,6 +300,23 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from values normalized through pydantic, so ordinary
+types that are not JSON -- a ``datetime`` or ``Decimal`` tool argument, a
+dataclass in ``tool_choice`` -- still fingerprint normally. If a value cannot be
+serialized even then, that step is not cached, and on retry it runs live rather
+than replaying an unverified entry.
+
+On the model path this is rarely confined to a single step: the causes are such a
+value in ``model_settings``, which is attached to every request, or in the message
+history, which every later request carries forward. Either one degrades all
+subsequent model steps the same way, leaving durable execution with nothing to
+replay, so the retry re-runs the agent at full cost. The
+``could not fingerprint model request`` warning names the step where this began.
+
+A tool call is fingerprinted from its name, arguments and call id alone, so
+neither of those causes reaches it. One that cannot be fingerprinted is reported
+as ``could not fingerprint tool call`` and costs only that call.
Review Comment:
"costs only that call" holds only when the live re-run returns the same content as the original attempt. The return lands in the next request's `ToolReturnPart.content`, which is part of the message history the following model step fingerprints, so a tool that reads mutable state or returns a timestamp makes every later model step miss too. Step 3 above already describes that cascade for divergence; a clause like "and, if its result differs, the model steps after it" keeps this sentence consistent with it.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -99,10 +114,18 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
def _digest(payload: Any) -> str:
- # No ``default=`` fallback: a non-JSON-serializable value must raise so the
- # callers degrade to an unverifiable (None) fingerprint instead of hashing
- # process-local reprs like ``<object at 0x...>`` that never match on retry.
- canonical = json.dumps(payload, sort_keys=True)
+ # Normalize through pydantic first. Values that are not JSON types but do have
+ # a deterministic pydantic serialization must not cost us a fingerprint: a
+ # ``datetime`` or ``Decimal`` tool argument (pydantic has already coerced tool
+ # arguments by the time they arrive) and a dataclass in ``tool_choice`` are
+ # ordinary, and hash identically on every attempt. Plain JSON values pass
+ # through unchanged, so fingerprints written by earlier versions still match.
+ #
+ # Still no ``default=`` fallback: a value pydantic cannot serialize either
+ # raises ``PydanticSerializationError`` (a ``ValueError``), so callers degrade
+ # to an unverifiable ``None`` fingerprint rather than hashing process-local
+ # reprs like ``<object at 0x...>`` that never match on retry.
+ canonical = json.dumps(to_jsonable_python(payload), sort_keys=True)
Review Comment:
`to_jsonable_python` turns a `set` or `frozenset` into a list in iteration order, and for `str` members that order depends on the process hash seed; `sort_keys=True` only sorts dict keys. Since every task attempt is a fresh process, a tool with a `set[str]` parameter now stores a digest on attempt 1 that attempt 2 never matches. Measured in breeze: six seeds gave six different digests for one `set[str]` payload, and a real pydantic-ai run stored `ba98e1f0...` then under another seed logged `reason='the conversation diverged from the previous attempt' step=1` and re-ran the tool live, while the same-seed control replayed. Before this PR that argument fingerprinted as `None`, so this is the one input where the new code costs a live re-run every retry rather than just declining to cache. A dataclass or `BaseModel` argument with a `set[str]` field, a nested set inside a list, and a tool that *returns* a set (via `ToolReturnPart.content` into the next model fingerprint) all behave the
same way; `set[int]` and single-member sets happen to be stable. A recursive canonicalizer that sorts set members by their JSON encoding before hashing (dump in `mode="python"` first so dataclasses unwrap but sets stay sets) gave one digest across all seeds and left plain payloads byte-identical when I tried it. No current test can catch this because every stability test compares two digests inside one process; a subprocess pair with different `PYTHONHASHSEED`, or asserting `_digest({"a": {"x", "y"}}) == _digest({"a": ["x", "y"]})`, would.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4048361827
##########
providers/common/ai/src/airflow/providers/common/ai/durable/caching_model.py:
##########
@@ -112,6 +114,11 @@ async def request(
)
response = await self.wrapped.request(messages, model_settings, model_request_parameters)
+ if fingerprint is None:
+ # Storing this would write an entry the guard above can never accept,
+ # once per step, each write rewriting the whole cache blob.
+ log.debug("Durable: not caching model response that cannot be verified on replay", step=step)
+ return response
Review Comment:
You're right, and I've reverted that half. "Executed" is the correct reading — `agent.py` renders these as `executed %d new steps` and `agent.rst` describes the line as replayed vs executed fresh, so suppressing the increment would print `executed 0 new steps` right after paying for a model call. That hides exactly the retry cost the new docs paragraph is there to surface.
Both counters now increment before the early return, and the two debug lines distinguish a step that was cached from one that ran without being persisted. The two tests that asserted `cached_model == 0` now assert `== 1`.
I kept the increment rather than adding a separate not-persisted counter — the summary line reads to me as being about cost, not storage accounting. Happy to add the second counter if you'd rather see both numbers.
That also settles the `save_tool_result` question I raised: under the "executed" reading its silent skip isn't an inconsistency, so there's nothing to fold in.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4048455744
##########
providers/common/ai/tests/unit/common/ai/durable/test_caching_model.py:
##########
@@ -158,6 +158,40 @@ async def test_legacy_entry_without_fingerprint_treated_as_miss(
assert result is sample_response
mock_model.request.assert_called_once()
+ @pytest.mark.asyncio
+ async def test_unverifiable_current_request_treated_as_miss(
+ self, mock_model, mock_storage, counter, sample_response
+ ):
+ stale = ModelResponse(parts=[TextPart(content="stale")])
+ mock_storage.load_model_response.return_value = (stale, None)
Review Comment:
I've Added `test_unverifiable_request_leaves_a_verifiable_entry_intact` plus the toolset equivalent:
a real stored fingerprint against a current request that can't be fingerprinted, asserting the step runs live, doesn't replay, and that `save_*` isn't called — so the usable entry survives for a later attempt instead of being clobbered with a `fingerprint=None` one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4150232903
##########
providers/common/ai/tests/unit/common/ai/durable/test_fingerprint.py:
##########
@@ -210,3 +237,339 @@ def test_arg_order_does_not_matter(self):
assert fingerprint_tool_call("t", {"a": 1, "b": 2}, "id1") == fingerprint_tool_call(
"t", {"b": 2, "a": 1}, "id1"
)
+
+
+class TestPydanticNativeValues:
+ """Values that are not JSON types but hash the same on every attempt must still fingerprint.
+
+ Tool arguments reach ``fingerprint_tool_call`` already coerced by pydantic, and
+ ``tool_choice`` accepts a dataclass while genuinely affecting the response, so it
+ cannot be stripped as transport-only. Since a step that cannot be fingerprinted is
+ no longer cached at all, refusing these values would stop an ordinary typed tool
+ from ever being cached.
+ """
+
+ def test_datetime_tool_argument_fingerprints(self):
+ when = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": when}, "id1") is not None
+
+ def test_decimal_tool_argument_fingerprints(self):
+ assert fingerprint_tool_call("t", {"amount": Decimal("10.5")}, "id1") is not None
+
+ def test_datetime_tool_argument_is_stable_and_distinguishing(self):
+ early = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ late = datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": early}, "id1") == fingerprint_tool_call(
+ "t", {"when": early}, "id1"
+ )
+ assert fingerprint_tool_call("t", {"when": early}, "id1") != fingerprint_tool_call(
+ "t", {"when": late}, "id1"
+ )
+
+ def test_tool_choice_dataclass_fingerprints(self):
+ fp = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["my_tool"])},
+ ModelRequestParameters(),
+ )
+
+ assert fp is not None
+
+ def test_tool_choice_dataclass_still_affects_the_fingerprint(self):
+ one = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["a"])},
+ ModelRequestParameters(),
+ )
+ other = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["b"])},
+ ModelRequestParameters(),
+ )
+
+ assert one is not None
+ assert one != other
+
+ def test_value_pydantic_cannot_serialize_still_returns_none(self):
+ """Normalization must not turn a genuinely unserializable value into a hash."""
+ assert fingerprint_tool_call("t", {"v": object()}, "id1") is None
+
+ def test_plain_payload_digest_is_unchanged_by_normalization(self):
+ """Fingerprints recorded before normalization must still match, so cached entries survive."""
+ payload = {"model": "m", "args": {"b": [1, True, None, "x", 2.5]}, "settings": None}
+ pre_normalization = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+ assert _digest(_render(payload)) == pre_normalization
+
+ def test_dict_keys_render_as_a_json_mode_dump_renders_them(self):
+ by_day = {datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}
+
+ assert _render({"series": by_day}) == {"series": {"2026-01-01": 1.5, "2026-01-02": 2.5}}
+ assert _render({1: "a", "b": 2}) == {"1": "a", "b": 2}
+
+ def test_keys_that_collide_once_rendered_are_refused(self):
+ """``1`` and ``"1"`` render alike, so hashing either payload could replay the other."""
+ assert fingerprint_tool_call("t", {"d": {1: "a", "1": "b"}}, "id1") is None
+
+ def test_bytes_that_are_not_utf8_render_as_base64(self):
+ """Decoding bytes as text would raise on an image and stop the tool from being cached."""
+ assert _render({"image": _PNG}) == {"image": base64.urlsafe_b64encode(_PNG).decode()}
+ assert fingerprint_tool_call("t", {"image": _PNG}, "id1") is not None
+
+
+def _json_mode_reference(model_identifier, messages, model_request_parameters):
+ """The fingerprint main computes: pydantic's json-mode dump, hashed as is.
+
+ For anything that is not a set, fingerprints must equal this, or entries stored
+ by an earlier version stop matching and the first retry after an upgrade re-runs
+ the whole agent.
+ """
+ dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
+ stripped = [
+ {
+ **{k: v for k, v in message.items() if k not in ("timestamp", "run_id", "conversation_id")},
+ "parts": [{k: v for k, v in part.items() if k != "timestamp"} for part in message["parts"]],
+ }
+ for message in dumped
+ ]
+ params = TypeAdapter(ModelRequestParameters).dump_python(model_request_parameters, mode="json")
+ payload = {"model": model_identifier, "messages": stripped, "settings": None, "params": params}
+ return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+
+def _with_tool_return(content):
+ return [
+ ModelRequest(parts=[UserPromptPart(content="go")]),
+ ModelResponse(parts=[ToolCallPart(tool_name="t", args={}, tool_call_id="c1")]),
+ ModelRequest(parts=[ToolReturnPart(tool_name="t", content=content, tool_call_id="c1")]),
+ ]
+
+
+class TestMessageHistoryMatchesJsonModeDump:
+ """The message history must hash exactly as pydantic's json-mode dump renders it.
+
+ Each case here is something a hand-written renderer gets wrong: raw bytes (not
+ valid UTF-8), dict keys that are not strings, ``NaN``, tuples, and fields whose
+ serializer only applies in JSON mode, such as ``InstructionPart.id``.
+ """
+
+ @pytest.mark.parametrize(
+ "messages",
+ [
+ pytest.param(
+ [
+ ModelRequest(
+ parts=[
+ UserPromptPart(content=["look", BinaryContent(data=_PNG, media_type="image/png")])
+ ]
+ )
+ ],
+ id="binary-content-in-prompt",
+ ),
+ pytest.param(_with_tool_return(_PNG), id="bytes-tool-return"),
+ pytest.param(
+ _with_tool_return({datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}),
+ id="date-keyed-tool-return",
+ ),
+ pytest.param(_with_tool_return({1: "a", "b": 2}), id="mixed-key-tool-return"),
+ pytest.param(_with_tool_return({"x": math.nan}), id="nan-tool-return"),
+ pytest.param(
+ _with_tool_return({"when": datetime.datetime(2026, 1, 1)}), id="datetime-tool-return"
+ ),
+ ],
+ )
+ def test_fingerprint_equals_the_json_mode_digest(self, messages):
+ fp = fingerprint_model_request("m", messages, None, ModelRequestParameters())
+
+ assert fp is not None
+ assert fp == _json_mode_reference("m", messages, ModelRequestParameters())
+
+ def test_request_parameters_hash_as_their_json_mode_dump(self):
+ """An agent's instructions reach the request parameters with a json-only serializer."""
+ seen = {}
+
+ class Spy(FunctionModel):
+ async def request(self, messages, model_settings, model_request_parameters):
+ seen.setdefault("messages", messages)
+ seen.setdefault("params", model_request_parameters)
+ return await super().request(messages, model_settings, model_request_parameters)
+
+ async def respond(messages, info):
+ return ModelResponse(parts=[TextPart(content="ok")])
+
+ Agent(Spy(respond), instructions="Be terse.").run_sync("hi")
+
+ fp = fingerprint_model_request("m", seen["messages"], None, seen["params"])
+
+ assert fp == _json_mode_reference("m", seen["messages"], seen["params"])
+
+ def test_tuple_parts_still_drop_part_timestamps(self):
+ """Message history passed as objects can hold its parts in a tuple."""
+ t1 = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ t2 = datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc)
+
+ def history(timestamp):
+ return [ModelRequest(parts=(UserPromptPart(content="hi", timestamp=timestamp),))]
+
+ assert fingerprint_model_request(
+ "m", history(t1), None, ModelRequestParameters()
+ ) == fingerprint_model_request("m", history(t2), None, ModelRequestParameters())
+
+ def test_set_in_a_tool_return_hashes_as_its_ordered_members(self):
+ assert fingerprint_model_request(
+ "m", _with_tool_return({"tags": {"b", "c", "a"}}), None, ModelRequestParameters()
+ ) == fingerprint_model_request(
+ "m", _with_tool_return({"tags": ["a", "b", "c"]}), None, ModelRequestParameters()
+ )
+
+
+class TestSetMemberOrdering:
+ """Sets must hash in a fixed order rather than the interpreter's iteration order.
+
+ A ``set[str]`` iterates in an order derived from the process hash seed, and every
+ task attempt runs in a fresh process. Hashing that order would produce a digest
+ the next attempt never reproduces, so the step would re-run live on every retry
+ -- worse than declining to cache it, which at least costs nothing extra.
+ """
+
+ def test_set_hashes_as_its_ordered_members(self):
+ assert _render({"tags": {"beta", "alpha"}}) == {"tags": ["alpha", "beta"]}
+
+ def test_set_matches_the_equivalent_list(self):
+ assert _render({"tags": {"alpha", "beta", "gamma"}}) == _render({"tags": ["alpha", "beta", "gamma"]})
+
+ def test_frozenset_matches_set(self):
+ assert _render({"tags": frozenset({"a", "b"})}) == _render({"tags": {"b", "a"}})
+
+ def test_different_members_still_produce_different_digests(self):
+ assert _render({"tags": {"a", "b"}}) != _render({"tags": {"a", "c"}})
+
+ def test_set_nested_inside_a_list(self):
+ assert _render({"filters": [{"z", "y"}]}) == {"filters": [["y", "z"]]}
+
+ def test_set_inside_a_dataclass_field(self):
+ @dataclasses.dataclass
+ class Filter:
+ tags: set
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_set_inside_a_basemodel_field(self):
+ class Filter(pydantic.BaseModel):
+ tags: set[str]
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_digest_is_stable_across_process_hash_seeds(self):
+ """The real proof: two fresh interpreters must agree, as two attempts would.
+
+ In-process comparisons cannot catch a hash-seed dependency, since one process
+ has one seed. The subprocess loads the module by path so it does not pay for
+ importing Airflow.
+ """
+ snippet = (
+ "import importlib.util;"
+ f"spec = importlib.util.spec_from_file_location('fp', r'{fingerprint_module.__file__}');"
+ "mod = importlib.util.module_from_spec(spec);"
+ "spec.loader.exec_module(mod);"
+ "print(mod._digest(mod._render({'tags': {'alpha', 'beta', 'gamma', 'delta'}})))"
+ )
+ digests = set()
+ for seed in ("0", "1", "2", "42"):
+ completed = subprocess.run(
+ [sys.executable, "-c", snippet],
+ capture_output=True,
+ text=True,
+ env={**os.environ, "PYTHONHASHSEED": seed, "PYTHONWARNINGS": "ignore"},
+ check=True,
+ )
+ digests.add(completed.stdout.strip().splitlines()[-1])
+
+ assert len(digests) == 1, f"digest depends on the hash seed: {digests}"
+
+ def test_set_returned_by_a_tool_is_stable_across_process_hash_seeds(self):
+ """The same proof through the message history, where a tool's set return lands."""
+ snippet = (
+ "import importlib.util;"
+ "from pydantic_ai.messages import ModelRequest, ToolReturnPart;"
+ "from pydantic_ai.models import ModelRequestParameters;"
+ f"spec = importlib.util.spec_from_file_location('fp', r'{fingerprint_module.__file__}');"
+ "mod = importlib.util.module_from_spec(spec);"
+ "spec.loader.exec_module(mod);"
+ "part = ToolReturnPart(tool_name='t', content={'alpha', 'beta', 'gamma', 'delta'}, tool_call_id='c1');"
+ "print(mod.fingerprint_model_request('m', [ModelRequest(parts=[part])], None, ModelRequestParameters()))"
+ )
+ digests = set()
+ for seed in ("0", "1", "2", "42"):
+ completed = subprocess.run(
+ [sys.executable, "-c", snippet],
+ capture_output=True,
+ text=True,
+ env={**os.environ, "PYTHONHASHSEED": seed, "PYTHONWARNINGS": "ignore"},
+ check=True,
+ )
+ digests.add(completed.stdout.strip().splitlines()[-1])
+
+ assert "None" not in digests
+ assert len(digests) == 1, f"digest depends on the hash seed: {digests}"
+
+
+class TestLazilyValidatedIterable:
+ """A lazily validated ``Iterable`` must be refused, not consumed.
+
+ ``to_jsonable_python`` drains an iterator to render it. Tool arguments are
+ fingerprinted before the tool runs, so draining one would hand the tool an
+ exhausted iterator and cache that wrong result under the fingerprint of the
+ full input. Declining to fingerprint leaves the step uncached instead.
+ """
+
+ def test_validator_iterator_argument_is_not_fingerprinted(self):
Review Comment:
Added these: nested iterators in a list, tuple, dict, deque, model field, dataclass field and extra, plus a generator in `ToolReturnPart.content` through `fingerprint_model_request`. The model gets every value now, where main gave it `[]`. `revealed_tool_names` is in the subprocess seed test too. I also reverted each safeguard one at a time, and every one of them fails the suite now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4234274441
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -123,44 +362,86 @@ def fingerprint_model_request(
output mode and schema, native tools, ...) so any change to what is sent
to the model invalidates the cached response.
- Returns ``None`` when the request cannot be serialized; ``None`` compares
- equal to ``None``, so requests that cannot be fingerprinted degrade to
- unverified positional replay rather than disabling caching.
+ Returns ``None`` when the request cannot be fingerprinted, which prevents the
+ step from being replayed or cached. Model settings, the tool definitions in the
+ request parameters and the message history are normally carried into every
+ later request, so such a value in any of them usually degrades every
+ subsequent model step of the run the same way. ``step`` is attached to the
+ warning so the log names where it began.
"""
try:
+ # Messages and parameters are hashed from pydantic's json-mode dump, as they
+ # always were, so stored fingerprints still match. The python-mode dumps only
+ # guide ``_order_sets``, which runs where a set or a non-string key needs it.
+ messages_guide = ModelMessagesTypeAdapter.dump_python(messages)
Review Comment:
You're right, that was a regression against main. Both guide dumps now catch `TypeError`, and when there is no guide the json-mode dump is hashed on its own, unordered, the way main does it. A frozenset of frozen dataclasses or models and a dict keyed by a frozen model all give main's digest again.
I ran the tests on pydantic 2.12.0 and 2.13.4, since the dict-key case takes a different path on each, and the test compares against main's digest so it holds on both. `_render` still raises for tool args, as you said. The comment above the dumps is corrected too.
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -107,6 +107,58 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Plain JSON arguments and settings are fingerprinted exactly as before. Anything
+else is fingerprinted from pydantic's JSON rendering, so ordinary types that are
+not JSON -- a ``datetime`` or ``Decimal`` tool argument, a dataclass in
+``tool_choice``, the bytes in a ``BinaryContent``, a dict keyed by date --
+fingerprint normally. Bytes in tool arguments and settings are rendered as base64,
+so binary data that is not valid UTF-8 fingerprints too; bytes inside a pydantic
+model follow that model's ``ser_json_bytes`` setting instead, which renders them as
+UTF-8 text by default. Because a tool call is fingerprinted from how its arguments
+render, a field excluded from serialization and a secret value, which renders
+masked, take no part in it.
+
+A step whose request cannot be fingerprinted is not cached, and on retry it runs
+live rather than replaying an unverified entry. That happens when pydantic cannot
Review Comment:
Agreed, "pydantic cannot render" was pointing at the wrong thing. With the fallback in, the paragraph now says this only applies to tool arguments and settings, where it was never fingerprinted anyway, and that in the message history these hash as they always did. It mentions dataclass instances as well as models.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -102,19 +159,201 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _check_guide(guide: Any) -> bool:
+ """
+ Refuse an iterator, and report whether the JSON rendering needs ``_order_sets``.
+
+ ``guide`` is pydantic's python-mode dump of the payload. It has the shape of the
+ json-mode rendering that is hashed -- the same keys, aliases, exclusions,
+ computed fields, extras and serializer output -- but keeps sets as sets and dict
+ keys as they are, and wraps an iterator in a ``SerializationIterator`` without
+ reading it. So this walk runs no user code, and nothing has been consumed yet.
+
+ Raises ``TypeError`` if the payload renders through an iterator anywhere.
+ Rendering it to JSON would consume it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into an iterator; a tool return that holds a generator
+ would reach the model empty. Returns ``True`` if the guide holds a set, or a dict
+ key that is not a string, since only those can make the JSON rendering depend on
+ the hash seed or lose a key.
+ """
+ needs_check = False
+ pending = [guide]
+ seen: set[int] = set()
+ while pending:
+ item = pending.pop()
+ if id(item) in seen:
+ # The dump shares this object, or left a cycle in place: walked already.
+ continue
+ seen.add(id(item))
+ children: Iterable[Any]
+ if isinstance(item, dict):
+ needs_check = needs_check or any(not isinstance(key, str) for key in item)
+ children = item.values()
+ elif isinstance(item, (list, tuple, deque)):
+ children = item
+ elif isinstance(item, (set, frozenset)):
+ needs_check = True
+ children = item
+ elif isinstance(item, Iterator):
+ raise TypeError(f"cannot fingerprint a value that renders through a {type(item).__name__}")
+ elif isinstance(item, enum.Enum):
+ # Python mode keeps an Enum member; JSON renders its value.
+ children = (item.value,)
+ else:
+ continue
+ pending.extend(child for child in children if type(child) not in _LEAF_TYPES)
+ return needs_check
+
+
+def _json_order(member: Any) -> str:
+ return json.dumps(member, sort_keys=True)
+
+
+def _template(value: Any) -> Any:
+ """
+ Say where the sets are inside one set member, or ``None`` if that cannot be said.
+
+ ``_LEAF`` for a member with no set inside it, ``("set", inner)`` for a set whose
+ members all have the template ``inner``, and ``("sequence", parts)`` for a tuple
+ with a set somewhere in it.
+ """
+ if isinstance(value, enum.Enum):
+ return _template(value.value)
+ if isinstance(value, (set, frozenset)):
+ inner = _member_template(value)
+ return None if inner is None else ("set", inner)
+ if isinstance(value, (list, tuple, deque)):
+ parts = tuple(_template(item) for item in value)
+ if None in parts:
+ return None
+ return _LEAF if all(part == _LEAF for part in parts) else ("sequence", parts)
+ if isinstance(value, dict):
+ # No set member dumps to a dict (python mode refuses a set of models), so
+ # nothing is known about one.
+ return None
+ return _LEAF
+
+
+def _member_template(members: Iterable[Any]) -> Any:
+ """Return the template every member of a set shares, or ``None`` if they differ."""
+ templates = {_template(member) for member in members}
+ if not templates:
+ return _LEAF
+ return templates.pop() if len(templates) == 1 else None
+
+
+def _apply_template(template: Any, rendered: Any) -> Any:
+ """Sort the lists that ``template`` puts a set at, in the rendering of one set member."""
+ if template == _LEAF or not isinstance(rendered, list):
+ return rendered
+ kind, inner = template
+ if kind == "set":
+ return sorted((_apply_template(inner, item) for item in rendered), key=_json_order)
+ if len(rendered) != len(inner):
+ return rendered
+ return [_apply_template(part, item) for part, item in zip(inner, rendered)]
+
+
+def _order_sets(guide: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is the json-mode rendering that is hashed, and ``guide`` the
+ python-mode dump of the same payload (see ``_check_guide``). They are walked
+ side by side: a dict or a sequence pairs up by position, because both dumps
+ keep the same order, and a list is sorted where the guide holds a set. A set's
+ members cannot pair up by position -- the dump builds a new set, which may
+ iterate in another order -- so the members' shared template says where any set
+ inside one of them sits (see ``_template``). Anything that does not line up is
+ kept exactly as rendered, so nothing that did not come from a set is reordered.
+
+ Raises ``TypeError`` where distinct dict keys render alike, such as ``1`` and
+ ``"1"``, or ``None`` and ``nan`` in a message dump: the guide then holds more keys
+ than the rendering, and hashing it would let two different payloads share a
+ digest. A dict whose keys are all strings cannot collide, so one that renders
+ fewer keys was reshaped by a serializer that applies only in JSON mode, and is
+ kept as rendered.
+ """
+ if isinstance(guide, enum.Enum):
+ # An Enum member renders as its value.
+ return _order_sets(guide.value, rendered)
+ if isinstance(guide, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(guide):
+ return rendered
+ template = _member_template(guide)
+ if template is None:
+ return rendered
+ return sorted((_apply_template(template, item) for item in rendered), key=_json_order)
Review Comment:
Took your approach. A list is now sorted only when the set, rendered on its own and ordered the same way, gives the same members. Your `tags`/`rows` swap gives different digests for `[1, 2]` and `[2, 1]` on both the tool path and the history, and matches main's digest.
For the other direction: when a set can't be found (mixed member shapes, a renamed set field, a reshaped one), the step is refused if the set's order follows the hash seed. If it doesn't, like a set of ints, it is kept as rendered so it still matches main.
One limit I couldn't close: a JSON-only serializer that puts a list with exactly the set's members in the set's place still gets sorted. I don't see a way to tell those apart. The docs now carry the `["z-first", "a"]` caveat.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -102,19 +159,201 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _check_guide(guide: Any) -> bool:
+ """
+ Refuse an iterator, and report whether the JSON rendering needs ``_order_sets``.
+
+ ``guide`` is pydantic's python-mode dump of the payload. It has the shape of the
+ json-mode rendering that is hashed -- the same keys, aliases, exclusions,
+ computed fields, extras and serializer output -- but keeps sets as sets and dict
+ keys as they are, and wraps an iterator in a ``SerializationIterator`` without
+ reading it. So this walk runs no user code, and nothing has been consumed yet.
+
+ Raises ``TypeError`` if the payload renders through an iterator anywhere.
+ Rendering it to JSON would consume it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into an iterator; a tool return that holds a generator
+ would reach the model empty. Returns ``True`` if the guide holds a set, or a dict
+ key that is not a string, since only those can make the JSON rendering depend on
+ the hash seed or lose a key.
+ """
+ needs_check = False
+ pending = [guide]
+ seen: set[int] = set()
+ while pending:
+ item = pending.pop()
+ if id(item) in seen:
+ # The dump shares this object, or left a cycle in place: walked already.
+ continue
+ seen.add(id(item))
+ children: Iterable[Any]
+ if isinstance(item, dict):
+ needs_check = needs_check or any(not isinstance(key, str) for key in item)
+ children = item.values()
+ elif isinstance(item, (list, tuple, deque)):
+ children = item
+ elif isinstance(item, (set, frozenset)):
+ needs_check = True
+ children = item
+ elif isinstance(item, Iterator):
+ raise TypeError(f"cannot fingerprint a value that renders through a {type(item).__name__}")
+ elif isinstance(item, enum.Enum):
+ # Python mode keeps an Enum member; JSON renders its value.
+ children = (item.value,)
+ else:
+ continue
+ pending.extend(child for child in children if type(child) not in _LEAF_TYPES)
+ return needs_check
+
+
+def _json_order(member: Any) -> str:
+ return json.dumps(member, sort_keys=True)
+
+
+def _template(value: Any) -> Any:
+ """
+ Say where the sets are inside one set member, or ``None`` if that cannot be said.
+
+ ``_LEAF`` for a member with no set inside it, ``("set", inner)`` for a set whose
+ members all have the template ``inner``, and ``("sequence", parts)`` for a tuple
+ with a set somewhere in it.
+ """
+ if isinstance(value, enum.Enum):
+ return _template(value.value)
+ if isinstance(value, (set, frozenset)):
+ inner = _member_template(value)
+ return None if inner is None else ("set", inner)
+ if isinstance(value, (list, tuple, deque)):
+ parts = tuple(_template(item) for item in value)
+ if None in parts:
+ return None
+ return _LEAF if all(part == _LEAF for part in parts) else ("sequence", parts)
+ if isinstance(value, dict):
+ # No set member dumps to a dict (python mode refuses a set of models), so
+ # nothing is known about one.
+ return None
+ return _LEAF
+
+
+def _member_template(members: Iterable[Any]) -> Any:
+ """Return the template every member of a set shares, or ``None`` if they differ."""
+ templates = {_template(member) for member in members}
+ if not templates:
+ return _LEAF
+ return templates.pop() if len(templates) == 1 else None
+
+
+def _apply_template(template: Any, rendered: Any) -> Any:
+ """Sort the lists that ``template`` puts a set at, in the rendering of one set member."""
+ if template == _LEAF or not isinstance(rendered, list):
+ return rendered
+ kind, inner = template
+ if kind == "set":
+ return sorted((_apply_template(inner, item) for item in rendered), key=_json_order)
+ if len(rendered) != len(inner):
+ return rendered
+ return [_apply_template(part, item) for part, item in zip(inner, rendered)]
+
+
+def _order_sets(guide: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is the json-mode rendering that is hashed, and ``guide`` the
+ python-mode dump of the same payload (see ``_check_guide``). They are walked
+ side by side: a dict or a sequence pairs up by position, because both dumps
+ keep the same order, and a list is sorted where the guide holds a set. A set's
+ members cannot pair up by position -- the dump builds a new set, which may
+ iterate in another order -- so the members' shared template says where any set
+ inside one of them sits (see ``_template``). Anything that does not line up is
+ kept exactly as rendered, so nothing that did not come from a set is reordered.
+
+ Raises ``TypeError`` where distinct dict keys render alike, such as ``1`` and
+ ``"1"``, or ``None`` and ``nan`` in a message dump: the guide then holds more keys
+ than the rendering, and hashing it would let two different payloads share a
+ digest. A dict whose keys are all strings cannot collide, so one that renders
+ fewer keys was reshaped by a serializer that applies only in JSON mode, and is
+ kept as rendered.
+ """
+ if isinstance(guide, enum.Enum):
+ # An Enum member renders as its value.
+ return _order_sets(guide.value, rendered)
+ if isinstance(guide, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(guide):
+ return rendered
+ template = _member_template(guide)
+ if template is None:
+ return rendered
+ return sorted((_apply_template(template, item) for item in rendered), key=_json_order)
+ if isinstance(guide, dict):
+ if not isinstance(rendered, dict):
+ return rendered
+ if len(rendered) != len(guide):
+ if len(rendered) < len(guide) and any(not isinstance(key, str) for key in guide):
+ raise TypeError("dict keys collide once rendered as JSON")
Review Comment:
Both fixed the way you suggested. Non-string keys are rendered on their own, and it only counts as a collision when that gives fewer keys than the guide. The filtered `dict[int, int]` hashes like main again and `{1: "a", "1": "b"}` is still refused.
The rendered keys also have to match the guide's keys in order now, so a reordered int-keyed dict is no longer paired with its neighbour. Your `{2: {"p", "q"}, 1: [...]}` pair is refused, and the same with a set of ints gives two different digests. I kept one extra refusal for `nan`/`inf` keys, because the message dump renders those as `None` and rendering the keys alone doesn't show it. The docstring line you quoted is rewritten.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4105722838
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -99,10 +114,18 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
def _digest(payload: Any) -> str:
Review Comment:
My last push turned a fail-open into silent data corruption, which is the worse of the two.
I reproduced it first: `TypeAdapter(Iterable[int]).validate_python([1, 2, 3])` gives a `ValidatorIterator`, `to_jsonable_python` drains it, and `sum()` afterwards is `0`. On the merge-base, `json.dumps` raised `TypeError` without advancing it, so the tool still got `6`. End to end with a real `Agent` + `FunctionModel` + `FunctionToolset` and `def total(values: Iterable[int])`, the tool saw `[]`, the agent answered `total=0`, and that `0` was cached under the fingerprint of `[1, 2, 3]`.
I went with refusing it rather than normalizing a deep copy. A copy only helps if copying a lazy iterator gives you an independent one, and that's exactly what you can't count on — whereas refusing lands on the not-cached path, where the step just re-runs live. `_canonical` raises `TypeError` for any `Iterator`, so a plain generator behaves the same way. I also checked the nested cases: an `Iterable[int]` field on a `BaseModel` or dataclass argument is refused too, with the values left intact for the tool.
Tests: `TestLazilyValidatedIterable` covers the refusal and checks the iterator is still readable afterwards, and `TestLazilyValidatedIterableArgument` in `test_replay_verification.py` is the real-agent regression test you asked for — the tool gets `[1, 2, 3]`, the answer is `total=6`, and nothing is written for that step.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-5849075533
Thanks @kaxil, this round was worth it. Instead of patching the renderer case by case, I took your larger point: pydantic's json-mode dump is the renderer again, exactly as on main, and the fingerprint only adds set ordering and iterator refusal on top. That covers the bytes, dict key, tuple, NaN and instruction cases together, and makes "stored fingerprints are unaffected" true again.
I checked every case you listed against main's digest on pydantic-ai 2.33 and 2.44, and reverted each part of the fix to make sure a test catches it. 130 durable tests pass. Replies are in each thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4042148056
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -63,9 +65,13 @@
# Settings that control transport, not response content. Excluded from the
# fingerprint: changing them should not invalidate a cached response, and some
-# (``timeout`` can be an ``httpx.Timeout``) are not JSON-serializable, which
-# would otherwise force the whole fingerprint to ``None`` and silently disable
-# replay verification for every step.
+# (``timeout`` can be an ``httpx.Timeout``) are not JSON-serializable.
+#
+# This frozenset is load-bearing. Model settings ride along with every request,
+# so a single non-JSON member fingerprints every model step as ``None`` -- which
+# now costs durable execution entirely for the run (nothing is cached, nothing
+# is replayed), not merely the verification of a replay. Any non-JSON setting
+# must be listed here or normalized before it reaches the fingerprint.
Review Comment:
Following up on the comment rewrite: two things violate this invariant today and they share one cause. `_digest` calls `json.dumps` with no `default=`, so it rejects pydantic-native values that would hash perfectly deterministically. `tool_choice` is one, since it accepts `ToolOrOutput`, a plain dataclass, and it can't go in the frozenset because it does affect the response. The other is on the tool side: `tool_args` reaches `fingerprint_tool_call` after pydantic has already coerced them, so an ordinary tool with a `datetime` or `Decimal` parameter fingerprints as `None` on every call, and after this PR that tool is never cached at all. Both reproduce on 2.23.0 (the floor) and 2.42.0. Passing the payload through `to_jsonable_python` before `json.dumps` covers both, leaves plain values byte-identical so in-flight entries don't shift, and still raises `PydanticSerializationError` on `extra_body=object()`, which the existing `except (TypeError, ValueError)` catches.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/caching_model.py:
##########
@@ -112,6 +114,11 @@ async def request(
)
response = await self.wrapped.request(messages, model_settings, model_request_parameters)
+ if fingerprint is None:
+ # Storing this would write an entry the guard above can never accept,
+ # once per step, each write rewriting the whole cache blob.
Review Comment:
The blob rewrite is specific to the `ObjectStorage` backend. `TaskStateStoreDurableStorage.save_model_response` sets a single key instead, and this class holds a `DurableStorageProtocol`, so the first half of the comment carries the point on its own. Separately, this debug line is the only place `step` gets attached: the `fingerprint.py` warning that the new docs paragraph points at as the marker has no `step`, so at production log levels there is nothing to say which step stopped fingerprinting. Threading `step` into that warning would be worth more than the debug line.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/caching_model.py:
##########
@@ -112,6 +114,11 @@ async def request(
)
response = await self.wrapped.request(messages, model_settings, model_request_parameters)
+ if fingerprint is None:
+ # Storing this would write an entry the guard above can never accept,
+ # once per step, each write rewriting the whole cache blob.
+ log.debug("Durable: not caching model response that cannot be verified on replay", step=step)
+ return response
Review Comment:
On your question about whether to keep the counter change: I think this half needs to go the other way. `cached_model` and `cached_tool` are rendered in `operators/agent.py` as `executed %d new steps`, and agent.rst describes that line as reporting steps replayed vs executed fresh, so the field means executed, not persisted. Skipping the increment here makes a run that replays step 0 and runs an unverifiable step 1 live print `executed 0 new steps` after paying for a model call, which is the retry cost your new docs paragraph is trying to make visible. The persisted reading doesn't hold anyway, since `save_tool_result` returns without writing on a non-serializable result and the counter still increments there. Either keep the increment and reword the log line, or add a separate not-persisted counter; the two new tests assert `cached_model == 0` so they'd move with it.
##########
providers/common/ai/tests/unit/common/ai/durable/test_caching_model.py:
##########
@@ -158,6 +158,40 @@ async def test_legacy_entry_without_fingerprint_treated_as_miss(
assert result is sample_response
mock_model.request.assert_called_once()
+ @pytest.mark.asyncio
+ async def test_unverifiable_current_request_treated_as_miss(
+ self, mock_model, mock_storage, counter, sample_response
+ ):
+ stale = ModelResponse(parts=[TextPart(content="stale")])
+ mock_storage.load_model_response.return_value = (stale, None)
Review Comment:
One case none of the four new tests reaches: a real stored fingerprint paired with a current request that cannot be fingerprinted. The read side was already safe there before this PR, since a real hash never equalled `None`, but the write side changed. That good entry used to get clobbered with a `fingerprint=None` one and now it survives for a later attempt to replay, which is the one new write behaviour with no coverage.
##########
providers/common/ai/docs/operators/agent.rst:
##########
@@ -300,6 +300,17 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+If a model request or tool call cannot be fingerprinted -- it carries a value
+that will not serialize to JSON -- that step is not cached, and on retry it runs
+live rather than replaying an unverified entry. This is rarely confined to a
+single step: the usual causes are a non-serializable value in ``model_settings``,
Review Comment:
Neither of these causes applies to a tool call, though the paragraph opens by covering both. `fingerprint_tool_call` only sees the tool name, args and call id, so `model_settings` and message history cannot reach it, and the warning it emits is `could not fingerprint tool call`, not the string quoted below. The dominant tool-call cause is an ordinary typed tool parameter, which is the `_digest` point on fingerprint.py. If that gets normalized this mostly resolves itself; otherwise the causes here need scoping to the model path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4150121294
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,19 +123,107 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _is_dataclass_instance(value: Any) -> bool:
+ return dataclasses.is_dataclass(value) and not isinstance(value, type)
+
+
+def _refuse_iterators(value: Any) -> None:
+ """
+ Raise ``TypeError`` if ``value`` holds an iterator anywhere inside it.
+
+ Rendering an iterator consumes it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into a ``ValidatorIterator``. Tool arguments are
+ fingerprinted before the tool runs, so rendering them would drain the tool's
+ own input: the tool would see an empty sequence, and that wrong result would be
+ cached under the fingerprint of the full one. Refusing degrades the step to the
+ not-cached path instead.
+ """
+ if value is None or isinstance(value, (str, bytes, bytearray, bool, int, float)):
+ return
+ if isinstance(value, Iterator):
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ children: Iterable[Any] = value.values()
+ elif isinstance(value, (list, tuple, set, frozenset)):
+ children = value
+ elif isinstance(value, BaseModel):
+ children = [getattr(value, name, None) for name in type(value).model_fields]
Review Comment:
Good catch, thanks. Extras are covered now. Instead of adding them to the walk, the fingerprint now checks pydantic's own python-mode dump of the arguments, which includes extras and wraps an iterator instead of reading it. Tool args are also rendered from a deep copy, and a `ValidatorIterator` can't be copied, so your `Iterable[int]` extra comes back `None` and still sums to 6. Added a test for it.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,19 +123,107 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _is_dataclass_instance(value: Any) -> bool:
+ return dataclasses.is_dataclass(value) and not isinstance(value, type)
+
+
+def _refuse_iterators(value: Any) -> None:
+ """
+ Raise ``TypeError`` if ``value`` holds an iterator anywhere inside it.
+
+ Rendering an iterator consumes it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into a ``ValidatorIterator``. Tool arguments are
+ fingerprinted before the tool runs, so rendering them would drain the tool's
+ own input: the tool would see an empty sequence, and that wrong result would be
+ cached under the fingerprint of the full one. Refusing degrades the step to the
+ not-cached path instead.
+ """
+ if value is None or isinstance(value, (str, bytes, bytearray, bool, int, float)):
+ return
+ if isinstance(value, Iterator):
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ children: Iterable[Any] = value.values()
+ elif isinstance(value, (list, tuple, set, frozenset)):
+ children = value
+ elif isinstance(value, BaseModel):
+ children = [getattr(value, name, None) for name in type(value).model_fields]
+ elif _is_dataclass_instance(value):
+ children = [getattr(value, field.name, None) for field in dataclasses.fields(value)]
+ else:
+ return
+ for child in children:
+ _refuse_iterators(child)
+
+
+def _order_sets(value: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is pydantic's JSON rendering of ``value`` and is otherwise kept as
+ is, so pydantic stays the only renderer: bytes, dates, dict keys, ``NaN`` and
+ custom serializers come out exactly as a json-mode dump renders them. The one
+ thing pydantic cannot do stably is order a set. It lists the members in
+ iteration order, which for strings follows the interpreter's hash seed, and
+ every task attempt is a fresh process, so a ``set[str]`` would hash differently
+ on each attempt and never replay. ``value`` is walked alongside ``rendered``
+ only to find those lists; a branch whose rendering does not line up with the
+ object, such as one with a custom serializer, is left exactly as rendered.
+ """
+ if isinstance(value, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(value):
+ return rendered
+ members = [_order_sets(member, item) for member, item in zip(value, rendered)]
+ return sorted(members, key=lambda member: json.dumps(member, sort_keys=True))
+ if isinstance(value, Mapping):
+ if not isinstance(rendered, dict):
+ return rendered
+ if len(rendered) != len(value):
+ # Distinct keys that render alike, such as 1 and "1": the digest could no
+ # longer tell those payloads apart, so refuse rather than hash either one.
+ raise TypeError("dict keys collide once rendered as JSON")
+ return {
+ key: _order_sets(item, rendered_item)
+ for (_, item), (key, rendered_item) in zip(value.items(), rendered.items())
+ }
+ if isinstance(value, (list, tuple)):
+ if not isinstance(rendered, list) or len(rendered) != len(value):
+ return rendered
+ return [_order_sets(item, rendered_item) for item, rendered_item in zip(value, rendered)]
+ if isinstance(rendered, dict) and (isinstance(value, BaseModel) or _is_dataclass_instance(value)):
+ return {
+ key: _order_sets(getattr(value, key), rendered_item) if hasattr(value, key) else rendered_item
Review Comment:
Fixed, but I didn't end up keeping the alias map. It worked for these three cases but kept missing others, like aliases a model's config puts on a nested stdlib dataclass. So the JSON rendering is now paired with pydantic's python-mode dump of the same value, which uses the same aliases and keeps sets as sets. Aliased, camelCase and root-model sets all give one digest across seeds now, and they're in the subprocess test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fingerprint durable agent steps that hold sets or values that are not JSON [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #71575:
URL: https://github.com/apache/airflow/pull/71575
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4048355986
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -63,9 +65,13 @@
# Settings that control transport, not response content. Excluded from the
# fingerprint: changing them should not invalidate a cached response, and some
-# (``timeout`` can be an ``httpx.Timeout``) are not JSON-serializable, which
-# would otherwise force the whole fingerprint to ``None`` and silently disable
-# replay verification for every step.
+# (``timeout`` can be an ``httpx.Timeout``) are not JSON-serializable.
+#
+# This frozenset is load-bearing. Model settings ride along with every request,
+# so a single non-JSON member fingerprints every model step as ``None`` -- which
+# now costs durable execution entirely for the run (nothing is cached, nothing
+# is replayed), not merely the verification of a replay. Any non-JSON setting
+# must be listed here or normalized before it reaches the fingerprint.
Review Comment:
Confirmed, and I'm glad you caught this one. It's a pre-existing bug, but my last commit turned it from "replays unverified" into "never cached at all", which is the worse of the two.
I reproduced all three on 2.29.0 before touching anything: a `datetime` argument, a `Decimal` argument, and `ToolOrOutput(function_tools=[...])` in `tool_choice` each come back `None` today. The tool one bothers me most — an ordinary typed parameter isn't an edge case, and it fails silently.
`to_jsonable_python` before `json.dumps`, as you suggested. I checked the three properties it leans on rather than taking them on trust: `PydanticSerializationError` subclasses `ValueError`, so the existing handler catches it unchanged; `extra_body=object()` still raises, so the unverifiable path is intact; and plain values round-trip byte-identically. That last one now has a test asserting `_digest` of a plain payload equals its pre-normalization sha256, so in-flight entries can't drift if pydantic changes later.
`timeout` as an `httpx.Timeout` still fails under pydantic, so I've kept `_TRANSPORT_ONLY_SETTINGS` and just reworded the comment to draw the line at what pydantic can't serialize rather than what isn't JSON.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4150235864
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -102,6 +102,33 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from pydantic's JSON rendering of each value, the same
+rendering a json-mode dump produces, so ordinary types that are not JSON -- a
+``datetime`` or ``Decimal`` tool argument, a dataclass in ``tool_choice``, the
+bytes in a ``BinaryContent``, a dict keyed by date -- fingerprint normally, and
+entries cached by an earlier version still match. The one adjustment is that the
+members of a ``set`` are ordered before hashing, so a set matches on a later
+attempt too. If a value cannot be rendered at all, that step is not cached, and on
+retry it runs live rather than replaying an unverified entry. A parameter annotated ``Iterable[...]`` is one such case:
+pydantic validates it lazily, and reading it in order to hash it would consume
+the input the tool itself has not read yet, so the step runs live instead of
+being cached.
+
+On the model path this is rarely confined to a single step: the causes are such a
+value in ``model_settings``, which is attached to every request, in the tool
+definitions the request carries, or in the message history, which every later
+request carries forward. Any one of them degrades all subsequent model steps the
+same way, leaving durable execution with nothing to
+replay, so the retry re-runs the agent at full cost. The
Review Comment:
Reworded along those lines. It now says it degrades every model step from that point on, so the retry re-runs from there. For settings and tool definitions that point is normally the first request, though callable settings or a tool's `prepare` can bring one in later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4109176855
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,11 +127,56 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _canonical(value: Any) -> Any:
+ """
+ Render ``value`` as JSON-safe data whose encoding is identical on every attempt.
+
+ ``to_jsonable_python`` on its own is not a safe fingerprint input, for two
+ reasons that both matter here because tool arguments arrive as live Python
+ objects that pydantic has already validated.
+
+ It renders a ``set`` in iteration order, and for string members that order
+ follows the interpreter's hash seed. Every task attempt is a fresh process, so
+ a ``set[str]`` argument would hash differently each time and never replay --
+ worse than declining to cache, because the step re-runs live on every retry.
+ Members are therefore ordered by their own JSON encoding.
+
+ It also *consumes* an iterator, and pydantic validates an ``Iterable[T]``
+ parameter lazily into a ``ValidatorIterator``. Normalizing the arguments would
+ drain the tool's own input before the tool ran, so the tool would see an empty
+ sequence and that wrong result would be cached under the fingerprint of the
+ full one. Such a value is refused instead, which degrades the step to the
+ not-cached path rather than corrupting it.
+ """
+ if value is None or isinstance(value, (str, bool, int, float)):
+ return value
+ if isinstance(value, Iterator):
+ # Hashing this means draining it, leaving nothing for the tool to read.
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ return {key: _canonical(item) for key, item in value.items()}
Review Comment:
Keys are passed through as-is here, and `json.dumps(sort_keys=True)` raises `TypeError` for anything but homogeneous `str` or `int` keys, where the json-mode dump used to stringify them. So a tool returning `dict[date, float]` (a series keyed by day), `dict[MyEnum, int]`, `dict[UUID, str]`, or a mixed `{1: "a", "b": 2}` lands in `ToolReturnPart.content` and every later model step fingerprints `None`; all of those digested on the merge-base (`495b81bd...` for the `date` case, `None` here). Rendering the key through pydantic, for example `next(iter(to_jsonable_python({key: None}, bytes_mode="base64")))` for non-`str` keys, gives digests identical to base for every key type I tried.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -85,9 +107,16 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
timestamps, part-level timestamps); user data such as tool arguments is
never recursed into, so an argument legitimately named ``run_id`` still
affects the fingerprint.
+
+ Raises ``TypeError`` if a message did not dump to a mapping. A python-mode dump
+ passes an object it does not recognize straight through, and a fingerprint that
Review Comment:
The reason given here is not what happens. Without the guard, an object passed straight through by the python-mode dump makes the next line raise `AttributeError`, which the `except (TypeError, ValueError, RecursionError)` below does not catch, so the task would fail rather than the fingerprint drifting between attempts. The only producer of `messages` is pydantic-ai's `Agent`, so `test_a_message_that_does_not_dump_to_a_mapping_returns_none` covers an input that cannot occur, and it passes for a different reason anyway: the `None` comes from `_canonical(object())` raising `PydanticSerializationError`, so the guard is never what returns it (swapping the raise for a passthrough leaves the suite green). Either drop the guard and that test, or keep them and state the real reason (and name `_strip_volatile` in the `except` comment).
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -102,6 +102,30 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from each value's canonical form, so ordinary types that
Review Comment:
Two things this paragraph promises are not true at head until the bytes and dict-key fixes land: "ordinary types that are not JSON fingerprint normally" fails for `bytes`, `BinaryContent` and non-string dict keys, and the older paragraph further down ("a `BinaryContent` from MCP tools means that step is skipped") now understates it, since the return stays in `ToolReturnPart.content` and every later model step misses too. Separately, the run-wide causes below list `model_settings` and message history, but `model_request_parameters` is hashed as well, so a tool definition with a non-serializable `json_schema_extra` has the same reach; "in `model_settings`, in the tool definitions the request carries, or in the message history" would cover it.
##########
providers/common/ai/tests/unit/common/ai/durable/test_fingerprint.py:
##########
@@ -210,3 +224,199 @@ def test_arg_order_does_not_matter(self):
assert fingerprint_tool_call("t", {"a": 1, "b": 2}, "id1") == fingerprint_tool_call(
"t", {"b": 2, "a": 1}, "id1"
)
+
+
+class TestPydanticNativeValues:
+ """Values that are not JSON types but hash the same on every attempt must still fingerprint.
+
+ Tool arguments reach ``fingerprint_tool_call`` already coerced by pydantic, and
+ ``tool_choice`` accepts a dataclass while genuinely affecting the response, so it
+ cannot be stripped as transport-only. Since a step that cannot be fingerprinted is
+ no longer cached at all, refusing these values would stop an ordinary typed tool
+ from ever being cached.
+ """
+
+ def test_datetime_tool_argument_fingerprints(self):
+ when = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": when}, "id1") is not None
+
+ def test_decimal_tool_argument_fingerprints(self):
+ assert fingerprint_tool_call("t", {"amount": Decimal("10.5")}, "id1") is not None
+
+ def test_datetime_tool_argument_is_stable_and_distinguishing(self):
+ early = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ late = datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": early}, "id1") == fingerprint_tool_call(
+ "t", {"when": early}, "id1"
+ )
+ assert fingerprint_tool_call("t", {"when": early}, "id1") != fingerprint_tool_call(
+ "t", {"when": late}, "id1"
+ )
+
+ def test_tool_choice_dataclass_fingerprints(self):
+ fp = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["my_tool"])},
+ ModelRequestParameters(),
+ )
+
+ assert fp is not None
+
+ def test_tool_choice_dataclass_still_affects_the_fingerprint(self):
+ one = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["a"])},
+ ModelRequestParameters(),
+ )
+ other = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["b"])},
+ ModelRequestParameters(),
+ )
+
+ assert one is not None
+ assert one != other
+
+ def test_value_pydantic_cannot_serialize_still_returns_none(self):
+ """Normalization must not turn a genuinely unserializable value into a hash."""
+ assert fingerprint_tool_call("t", {"v": object()}, "id1") is None
+
+ def test_plain_payload_digest_is_unchanged_by_normalization(self):
Review Comment:
This pins a hand-built dict, not the message dump path, so the `mode="python"` switch is invisible to the suite: with both `dump_python` calls reverted to `mode="json"` all 68 tests still pass, and nothing here mentions `BinaryContent` or `bytes`. A golden digest recorded from main for a `ModelRequest` carrying a non-UTF-8 `BinaryContent`, and one for a `bytes` `ToolReturnPart`, would fail without the `bytes_mode` fix; and a `set` inside `ToolReturnPart.content` asserted stable across hash seeds would couple the mode switch to the case it was made for.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -85,9 +107,16 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
timestamps, part-level timestamps); user data such as tool arguments is
never recursed into, so an argument legitimately named ``run_id`` still
affects the fingerprint.
+
+ Raises ``TypeError`` if a message did not dump to a mapping. A python-mode dump
+ passes an object it does not recognize straight through, and a fingerprint that
+ cannot strip the volatile fields would change on every attempt, so the caller
+ degrades to an unverifiable ``None`` instead.
"""
stripped = []
for message in messages_dump:
+ if not isinstance(message, Mapping):
+ raise TypeError(f"expected a dumped message mapping, got {type(message).__name__}")
cleaned = {k: v for k, v in message.items() if k not in _VOLATILE_MESSAGE_KEYS}
if isinstance(cleaned.get("parts"), list):
Review Comment:
One more thing the mode switch exposes: the json-mode dump rendered `parts` as a list whatever the input, but python mode preserves a tuple, and this `list` check then skips the timestamp strip. `AgentOperator(message_history=[ModelRequest(parts=(UserPromptPart(...),))])` reaches here with a tuple because `_resolve_message_history` runs `validate_python` on object input, which keeps instances as-is (I checked: `parts` is still a `tuple` after validation). Each attempt then gets a fresh part timestamp into the digest, so every model step re-runs live with a "does not match the current request" warning that reads as if the agent changed, where the merge-base was stable across attempts. `isinstance(..., (list, tuple))` fixes it; a test with tuple parts and two different part timestamps asserting equal fingerprints would cover it. The XCom round-trip yields lists, so that path is unaffected.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -119,13 +195,21 @@ def fingerprint_model_request(
output mode and schema, native tools, ...) so any change to what is sent
to the model invalidates the cached response.
- Returns ``None`` when the request cannot be serialized; ``None`` compares
- equal to ``None``, so requests that cannot be fingerprinted degrade to
- unverified positional replay rather than disabling caching.
+ Returns ``None`` when the request cannot be serialized even through pydantic,
+ which prevents the step from being replayed or cached. Because model settings
+ and message history are carried into every later request, such a value in
+ either usually degrades every subsequent model step of the run the same way.
+ ``step`` is attached to that warning so the log names where it began.
"""
try:
- dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
- params = _MODEL_REQUEST_PARAMETERS_ADAPTER.dump_python(model_request_parameters, mode="json")
+ # ``mode="python"``, not ``mode="json"``: a json-mode dump renders a set as
+ # a list in iteration order, and a tool that returned a set puts one in the
+ # message history, where it would reach the hash already unstably ordered.
+ # Python mode leaves it a set for ``_canonical`` to order. For values that
+ # are not sets the two modes produce the same digest, so stored
+ # fingerprints are unaffected.
+ dumped = ModelMessagesTypeAdapter.dump_python(messages)
Review Comment:
`ModelMessagesTypeAdapter` sets `ser_json_bytes="base64"`, but that config only applies to json-mode dumps. With `mode="python"` the `bytes` inside `BinaryContent.data` reach `_canonical` raw, and `to_jsonable_python` decodes bytes as utf-8 by default. A PNG raises `UnicodeDecodeError` there, so the request fingerprints as `None`.
The attachment stays in the message history, so once a prompt contains an image the model steps after it are never cached for that run. The merge-base handled this case: a PNG in a `UserPromptPart` digested to `956c3851...` there and gives `None` here. I ran it end to end with `CachingModel` and `CachingToolset`: attempt 1 wrote no model entry, and attempt 2 reported `replayed_model=0 replayed_tool=0`. The same happens for a `bytes` or `BinaryContent` tool return, a `FilePart` in a response, and bytes in `provider_details`. Bytes that happen to decode as utf-8 do not raise; they produce a different digest than main, so those stored entries mismatch without any warning.
`to_jsonable_python(value, bytes_mode="base64")` on line 173 produces the same `iVBORw==` string the json-mode dump did, and in my probes it restored the base digest for every carrier above. A test that pins `_digest` of a history containing a `BinaryContent` to its pre-PR json-mode digest would guard it, in the same shape as `test_plain_payload_digest_is_unchanged_by_normalization`.
The larger issue is that `_canonical` now has to reproduce pydantic's json-mode rules by hand. Bytes are one case; dict keys (next comment) and `nan`/`inf` (rendered as `NaN`/`Infinity` here, `null` in json mode) are two more. If `_canonical` only did what pydantic cannot do stably, which is ordering sets, refusing iterators and expanding dataclasses, and passed every leaf and every key through `to_jsonable_python(bytes_mode="base64")`, pydantic would stay the only renderer. A golden-digest fixture covering bytes, a `datetime`, a `Decimal`, a set and a dict with non-string keys would then catch any future drift.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -119,13 +195,21 @@ def fingerprint_model_request(
output mode and schema, native tools, ...) so any change to what is sent
to the model invalidates the cached response.
- Returns ``None`` when the request cannot be serialized; ``None`` compares
- equal to ``None``, so requests that cannot be fingerprinted degrade to
- unverified positional replay rather than disabling caching.
+ Returns ``None`` when the request cannot be serialized even through pydantic,
+ which prevents the step from being replayed or cached. Because model settings
+ and message history are carried into every later request, such a value in
+ either usually degrades every subsequent model step of the run the same way.
+ ``step`` is attached to that warning so the log names where it began.
"""
try:
- dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
- params = _MODEL_REQUEST_PARAMETERS_ADAPTER.dump_python(model_request_parameters, mode="json")
+ # ``mode="python"``, not ``mode="json"``: a json-mode dump renders a set as
+ # a list in iteration order, and a tool that returned a set puts one in the
+ # message history, where it would reach the hash already unstably ordered.
+ # Python mode leaves it a set for ``_canonical`` to order. For values that
+ # are not sets the two modes produce the same digest, so stored
Review Comment:
"Stored fingerprints are unaffected" is not true for any agent with instructions, independent of the bytes case. `InstructionPart.id` is an `InstructionId` dataclass with a `PlainSerializer(when_used="json")` that renders `"agent"` / `"toolset:x:name"` only in json mode; python mode dumps it as a nested dict, so a real `Agent(instructions="Be terse.")` digests `d7694c37...` on main and `a4f1c4aa...` here, while an uninstructed agent matches. The same applies to `nan`/`inf` (`null` vs `NaN`) and to any user `BaseModel` field with a `when_used="json"` serializer. The effect is bounded: the first retry that straddles the upgrade misses every cached model step and re-runs the agent once, with the "model, prompt, message history, settings, or tools changed" reason, and is stable afterwards. Either restore json-mode rendering for these (handing leaves to `to_jsonable_python` in json semantics gets most of the way) or replace the "unaffected" claims here, in `_digest`, and in the docs wi
th a statement of the one-time re-run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r3972084886
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -34,9 +34,8 @@
Fields that pydantic-ai regenerates on every attempt (message-level
``timestamp``/``run_id``/``conversation_id`` and part-level ``timestamp``)
are excluded from the fingerprint. Requests that cannot be serialized to
-JSON fingerprint as ``None``, which degrades that step to unverified
-positional replay (the pre-fingerprint behavior) rather than disabling
-caching.
+JSON fingerprint as ``None`` and re-run live instead of replaying without
Review Comment:
Thanks @kaxil — all three addressed in `98004ea265`.
**1. Skipping the save when `fingerprint is None`** — done, in both `caching_model.py` and `caching_toolset.py`. The entry is unreachable exactly as you describe: the guard needs a non-`None` current fingerprint, and a real digest never equals a stored `None`, so the write only bought a dead entry per step plus a full rewrite of the cache blob each time. The step now returns before the save.
One consequence worth noting: the warning below the comparison is now reachable only for entries that predate fingerprinting, since a `None` fingerprint is no longer something we can write. I also stopped incrementing `cached_model` / `cached_tool` on that path — otherwise the INFO summary reports steps as cached when nothing was persisted.
Happy to drop that part if you'd rather keep this commit strictly to the dead-write fix.
**2. "That step" undersells it** — agreed, and it matches what I see putting those cases through `fingerprint_model_request`. The docs now state that the realistic causes — a non-JSON value in `model_settings`, which accompanies every request, or in the message history, which every later request carries forward — degrade every subsequent model step, leaving durable execution with nothing to replay and a retry that re-runs the agent at full cost. It also points at the `could not fingerprint model request` warning as the marker for where it began.
**3. The `_TRANSPORT_ONLY_SETTINGS` comment** — updated to the current consequence: the frozenset is load-bearing, because one non-JSON member now costs caching and replay for the whole run rather than just the verification. I carried the same correction into the module docstring, `fingerprint_model_request`'s docstring, both `log.warning` strings (they still read "this step will execute live on retry"), and the docstring on `test_httpx_timeout_does_not_disable_fingerprint`, which had the same stale framing.
**Tests** — the two existing unverifiable-step tests now assert that nothing is written, plus one new test per wrapper for the fresh-cache case. All four fail without this change and pass with it.
One adjacent thing I noticed but deliberately left out of scope: `DurableStorage.save_tool_result` skips the write for a non-serializable result, but the caller still counts it as cached, so the same counter skew exists on that path. Want that folded in here, or kept separate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4105742636
##########
providers/common/ai/docs/operators/agent.rst:
##########
@@ -300,6 +300,23 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from values normalized through pydantic, so ordinary
Review Comment:
Rebased onto main — the three paragraphs now live in `durable_execution.rst`, right after the "cached steps are deleted" line, and `agent.rst` is byte-identical to main, so #73523 is untouched.
On the wording: the canonical form landed, so "still fingerprint normally" is true as written now. I made it name `set` explicitly as one of the types that fingerprints and still matches on a later attempt, and it also calls out a lazily validated `Iterable` as the one ordinary annotation that's deliberately left uncached.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4112452294
##########
providers/common/ai/tests/unit/common/ai/durable/test_fingerprint.py:
##########
@@ -210,3 +224,199 @@ def test_arg_order_does_not_matter(self):
assert fingerprint_tool_call("t", {"a": 1, "b": 2}, "id1") == fingerprint_tool_call(
"t", {"b": 2, "a": 1}, "id1"
)
+
+
+class TestPydanticNativeValues:
+ """Values that are not JSON types but hash the same on every attempt must still fingerprint.
+
+ Tool arguments reach ``fingerprint_tool_call`` already coerced by pydantic, and
+ ``tool_choice`` accepts a dataclass while genuinely affecting the response, so it
+ cannot be stripped as transport-only. Since a step that cannot be fingerprinted is
+ no longer cached at all, refusing these values would stop an ordinary typed tool
+ from ever being cached.
+ """
+
+ def test_datetime_tool_argument_fingerprints(self):
+ when = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": when}, "id1") is not None
+
+ def test_decimal_tool_argument_fingerprints(self):
+ assert fingerprint_tool_call("t", {"amount": Decimal("10.5")}, "id1") is not None
+
+ def test_datetime_tool_argument_is_stable_and_distinguishing(self):
+ early = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ late = datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": early}, "id1") == fingerprint_tool_call(
+ "t", {"when": early}, "id1"
+ )
+ assert fingerprint_tool_call("t", {"when": early}, "id1") != fingerprint_tool_call(
+ "t", {"when": late}, "id1"
+ )
+
+ def test_tool_choice_dataclass_fingerprints(self):
+ fp = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["my_tool"])},
+ ModelRequestParameters(),
+ )
+
+ assert fp is not None
+
+ def test_tool_choice_dataclass_still_affects_the_fingerprint(self):
+ one = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["a"])},
+ ModelRequestParameters(),
+ )
+ other = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["b"])},
+ ModelRequestParameters(),
+ )
+
+ assert one is not None
+ assert one != other
+
+ def test_value_pydantic_cannot_serialize_still_returns_none(self):
+ """Normalization must not turn a genuinely unserializable value into a hash."""
+ assert fingerprint_tool_call("t", {"v": object()}, "id1") is None
+
+ def test_plain_payload_digest_is_unchanged_by_normalization(self):
Review Comment:
Added a test helper that hashes the json-mode dump the way main does, and cases that must equal it: non-UTF-8 `BinaryContent`, a bytes tool return, date and mixed keys, NaN, a datetime, and an agent's `instruction_parts`. Reverting either dump to python mode now fails them. I compute main's digest in the test rather than hard-coding it, since a fixed digest would break whenever pydantic-ai adds a field. There's also a subprocess test for a set inside `ToolReturnPart.content` across hash seeds.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -119,13 +195,21 @@ def fingerprint_model_request(
output mode and schema, native tools, ...) so any change to what is sent
to the model invalidates the cached response.
- Returns ``None`` when the request cannot be serialized; ``None`` compares
- equal to ``None``, so requests that cannot be fingerprinted degrade to
- unverified positional replay rather than disabling caching.
+ Returns ``None`` when the request cannot be serialized even through pydantic,
+ which prevents the step from being replayed or cached. Because model settings
+ and message history are carried into every later request, such a value in
+ either usually degrades every subsequent model step of the run the same way.
+ ``step`` is attached to that warning so the log names where it began.
"""
try:
- dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
- params = _MODEL_REQUEST_PARAMETERS_ADAPTER.dump_python(model_request_parameters, mode="json")
+ # ``mode="python"``, not ``mode="json"``: a json-mode dump renders a set as
+ # a list in iteration order, and a tool that returned a set puts one in the
+ # message history, where it would reach the hash already unstably ordered.
+ # Python mode leaves it a set for ``_canonical`` to order. For values that
+ # are not sets the two modes produce the same digest, so stored
Review Comment:
With the json-mode dump back, that claim holds again. On 2.44, `Agent(instructions="Be terse.")` gives the same digest as main (it drifted at the previous head), and NaN renders as `null` again. Checked on 2.33 and 2.44.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4048366758
##########
providers/common/ai/src/airflow/providers/common/ai/durable/caching_model.py:
##########
@@ -112,6 +114,11 @@ async def request(
)
response = await self.wrapped.request(messages, model_settings, model_request_parameters)
+ if fingerprint is None:
+ # Storing this would write an entry the guard above can never accept,
+ # once per step, each write rewriting the whole cache blob.
Review Comment:
Both done. You're right that the blob rewrite is an `ObjectStorage` detail and this class only holds a `DurableStorageProtocol`, with `TaskStateStoreDurableStorage` setting a single key — so that half is gone. What's left ("an entry the guard can never accept") makes the point without naming a backend.
Agreed the marker was the more useful half. `fingerprint_model_request` and `fingerprint_tool_call` now both take a keyword-only `step` and attach it to their warnings, so the string the docs point at names the step at production log levels instead of only the debug line carrying it. It defaults to `None`, so existing calls still work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on PR #71575:
URL: https://github.com/apache/airflow/pull/71575#issuecomment-5734852067
Thanks @kaxil — all five are addressed, split one commit per point: `e1de6303` (the `_digest` normalization), `b1fd367b` (counter), `d00158f8` (step in the warnings), `1fd22218` (the missing coverage case), `f85947d4` (the docs paragraph). The `_digest` one was the real find: it's a pre-existing bug, but my previous commit made it worse rather than introducing it, so I've treated it as the root fix and let the rest follow from there.
I reproduced each case before changing anything, on pydantic-ai 2.29.0:
| case | fingerprint today |
|---|---|
| `datetime` tool argument | `None` |
| `Decimal` tool argument | `None` |
| `ToolOrOutput` in `tool_choice` | `None` |
`_digest` now runs the payload through `to_jsonable_python` before `json.dumps`. I checked the three properties that make that safe instead of assuming them: `PydanticSerializationError` subclasses `ValueError`, so the existing `except (TypeError, ValueError)` catches it unchanged; `object()` still raises, so an unverifiable request still fingerprints as `None`; and plain JSON values come back byte-identical, so entries already written keep matching. There's a test pinning that last one to the pre-normalization sha256, so a future pydantic change can't quietly shift stored fingerprints.
Worth flagging: `httpx.Timeout` still fails under pydantic too, so `_TRANSPORT_ONLY_SETTINGS` is still load-bearing. I've narrowed that comment to draw the line at what pydantic can't serialize, since merely non-JSON values are fine now.
No new dependency — this provider already imports from `pydantic_core` in `utils/tool_definition.py`.
Tests: 98 in `providers/common/ai/tests/unit/common/ai/durable`, nine of them new here. Reverting just the `to_jsonable_python` call fails exactly the five pydantic-native tests while the stability and `object()` tests still pass, which is the signature I was after — the change only moves what it's meant to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "YAshhh29 (via GitHub)" <gi...@apache.org>.
YAshhh29 commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4105750067
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -99,10 +114,18 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
def _digest(payload: Any) -> str:
- # No ``default=`` fallback: a non-JSON-serializable value must raise so the
- # callers degrade to an unverifiable (None) fingerprint instead of hashing
- # process-local reprs like ``<object at 0x...>`` that never match on retry.
- canonical = json.dumps(payload, sort_keys=True)
+ # Normalize through pydantic first. Values that are not JSON types but do have
+ # a deterministic pydantic serialization must not cost us a fingerprint: a
+ # ``datetime`` or ``Decimal`` tool argument (pydantic has already coerced tool
+ # arguments by the time they arrive) and a dataclass in ``tool_choice`` are
+ # ordinary, and hash identically on every attempt. Plain JSON values pass
+ # through unchanged, so fingerprints written by earlier versions still match.
+ #
+ # Still no ``default=`` fallback: a value pydantic cannot serialize either
+ # raises ``PydanticSerializationError`` (a ``ValueError``), so callers degrade
+ # to an unverifiable ``None`` fingerprint rather than hashing process-local
+ # reprs like ``<object at 0x...>`` that never match on retry.
+ canonical = json.dumps(to_jsonable_python(payload), sort_keys=True)
Review Comment:
Confirmed, including the parts you scoped out: `set[int]` and single-member sets are stable, `set[str]` isn't. Six hash seeds gave me five distinct digests for one `set[str]` payload.
`_canonical` now orders set and frozenset members by their own JSON encoding — one digest across every seed I tried, and plain payloads unchanged. It recurses, so a set nested in a list and a `set[str]` field on a dataclass or `BaseModel` are covered. `BaseModel` goes through `model_dump(mode="python")`, so its nested sets arrive as sets and get ordered rather than being flattened on the way in.
The tool-return case needed one more change: the message dump was `mode="json"`, which turned a returned set into a list before `_canonical` ever saw it (four digests across seeds). Messages and request parameters now dump in `mode="python"`. I checked that for anything that isn't a set, the two modes give the same digest — including on a 451-message history — so stored fingerprints don't move.
On tests, you were right that in-process comparisons can't catch this, so there are both kinds: deterministic ones that assert the canonical form directly, and a subprocess test across four `PYTHONHASHSEED` values. The subprocess one earned its keep — with the ordering switched off, several in-process set assertions still passed by luck, while it failed with three distinct digests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Do not replay cached AI steps that cannot be verified [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #71575:
URL: https://github.com/apache/airflow/pull/71575#discussion_r4127938893
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,19 +123,107 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _is_dataclass_instance(value: Any) -> bool:
+ return dataclasses.is_dataclass(value) and not isinstance(value, type)
+
+
+def _refuse_iterators(value: Any) -> None:
+ """
+ Raise ``TypeError`` if ``value`` holds an iterator anywhere inside it.
+
+ Rendering an iterator consumes it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into a ``ValidatorIterator``. Tool arguments are
+ fingerprinted before the tool runs, so rendering them would drain the tool's
+ own input: the tool would see an empty sequence, and that wrong result would be
+ cached under the fingerprint of the full one. Refusing degrades the step to the
+ not-cached path instead.
+ """
+ if value is None or isinstance(value, (str, bytes, bytearray, bool, int, float)):
+ return
+ if isinstance(value, Iterator):
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ children: Iterable[Any] = value.values()
+ elif isinstance(value, (list, tuple, set, frozenset)):
+ children = value
+ elif isinstance(value, BaseModel):
+ children = [getattr(value, name, None) for name in type(value).model_fields]
Review Comment:
This walks `model_fields` only, so a model's extras never reach the guard, but `to_jsonable_python` does render them. A tool argument typed as a model with `extra="allow"` and `__pydantic_extra__: dict[str, Iterable[int]]` validates each extra into a `ValidatorIterator`, even from JSON args, and the render drains it before the tool runs. I ran it through a real `Agent` with `FunctionModel` and `FunctionToolset`, calling a tool that sums the extra: the fingerprint came back non-`None` and the tool returned 0 for `[1, 2, 3]`. That is the drain-then-cache-the-wrong-result path the `Iterable[T]` refusal exists for. Adding `(value.model_extra or {}).values()` to the children here closes it: the same argument then fingerprints as `None` and the extra still sums to 6.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,19 +123,107 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _is_dataclass_instance(value: Any) -> bool:
+ return dataclasses.is_dataclass(value) and not isinstance(value, type)
+
+
+def _refuse_iterators(value: Any) -> None:
+ """
+ Raise ``TypeError`` if ``value`` holds an iterator anywhere inside it.
+
+ Rendering an iterator consumes it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into a ``ValidatorIterator``. Tool arguments are
+ fingerprinted before the tool runs, so rendering them would drain the tool's
+ own input: the tool would see an empty sequence, and that wrong result would be
+ cached under the fingerprint of the full one. Refusing degrades the step to the
+ not-cached path instead.
+ """
+ if value is None or isinstance(value, (str, bytes, bytearray, bool, int, float)):
+ return
+ if isinstance(value, Iterator):
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ children: Iterable[Any] = value.values()
+ elif isinstance(value, (list, tuple, set, frozenset)):
+ children = value
+ elif isinstance(value, BaseModel):
+ children = [getattr(value, name, None) for name in type(value).model_fields]
+ elif _is_dataclass_instance(value):
+ children = [getattr(value, field.name, None) for field in dataclasses.fields(value)]
+ else:
+ return
+ for child in children:
+ _refuse_iterators(child)
+
+
+def _order_sets(value: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is pydantic's JSON rendering of ``value`` and is otherwise kept as
+ is, so pydantic stays the only renderer: bytes, dates, dict keys, ``NaN`` and
+ custom serializers come out exactly as a json-mode dump renders them. The one
+ thing pydantic cannot do stably is order a set. It lists the members in
+ iteration order, which for strings follows the interpreter's hash seed, and
+ every task attempt is a fresh process, so a ``set[str]`` would hash differently
+ on each attempt and never replay. ``value`` is walked alongside ``rendered``
+ only to find those lists; a branch whose rendering does not line up with the
+ object, such as one with a custom serializer, is left exactly as rendered.
+ """
+ if isinstance(value, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(value):
+ return rendered
+ members = [_order_sets(member, item) for member, item in zip(value, rendered)]
+ return sorted(members, key=lambda member: json.dumps(member, sort_keys=True))
+ if isinstance(value, Mapping):
+ if not isinstance(rendered, dict):
+ return rendered
+ if len(rendered) != len(value):
+ # Distinct keys that render alike, such as 1 and "1": the digest could no
+ # longer tell those payloads apart, so refuse rather than hash either one.
+ raise TypeError("dict keys collide once rendered as JSON")
+ return {
+ key: _order_sets(item, rendered_item)
+ for (_, item), (key, rendered_item) in zip(value.items(), rendered.items())
+ }
+ if isinstance(value, (list, tuple)):
+ if not isinstance(rendered, list) or len(rendered) != len(value):
+ return rendered
+ return [_order_sets(item, rendered_item) for item, rendered_item in zip(value, rendered)]
+ if isinstance(rendered, dict) and (isinstance(value, BaseModel) or _is_dataclass_instance(value)):
+ return {
+ key: _order_sets(getattr(value, key), rendered_item) if hasattr(value, key) else rendered_item
Review Comment:
Matching rendered keys back to attributes by name misses two shapes, and their sets keep hash-seed order. An aliased field renders under its alias, so `hasattr` fails. That covers `tags: set[str] = Field(alias="labels")` in a tool argument, and a camelCase model (`alias_generator=to_camel, serialize_by_alias=True`) returned into `ToolReturnPart.content`, where every later model step then misses on retry with the "changed" reason and no "could not fingerprint" warning. A `RootModel[set[str]]` renders straight to a list and never reaches this branch. Three seeds gave three digests for each. Mapping the rendered key back through `{f.serialization_alias or f.alias or name: name for name, f in type(value).__pydantic_fields__.items()}` (plain dataclasses fall back to the name) and recursing into `value.root` for a `RootModel` made all three agree across seeds when I tried it.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -98,19 +123,107 @@ def _strip_volatile(messages_dump: list[dict[str, Any]]) -> list[dict[str, Any]]
return stripped
+def _is_dataclass_instance(value: Any) -> bool:
+ return dataclasses.is_dataclass(value) and not isinstance(value, type)
+
+
+def _refuse_iterators(value: Any) -> None:
+ """
+ Raise ``TypeError`` if ``value`` holds an iterator anywhere inside it.
+
+ Rendering an iterator consumes it, and pydantic validates an ``Iterable[T]``
+ tool parameter lazily into a ``ValidatorIterator``. Tool arguments are
+ fingerprinted before the tool runs, so rendering them would drain the tool's
+ own input: the tool would see an empty sequence, and that wrong result would be
+ cached under the fingerprint of the full one. Refusing degrades the step to the
+ not-cached path instead.
+ """
+ if value is None or isinstance(value, (str, bytes, bytearray, bool, int, float)):
+ return
+ if isinstance(value, Iterator):
+ raise TypeError(f"cannot fingerprint {type(value).__name__} without consuming it")
+ if isinstance(value, Mapping):
+ children: Iterable[Any] = value.values()
+ elif isinstance(value, (list, tuple, set, frozenset)):
+ children = value
+ elif isinstance(value, BaseModel):
+ children = [getattr(value, name, None) for name in type(value).model_fields]
+ elif _is_dataclass_instance(value):
+ children = [getattr(value, field.name, None) for field in dataclasses.fields(value)]
+ else:
+ return
+ for child in children:
+ _refuse_iterators(child)
+
+
+def _order_sets(value: Any, rendered: Any) -> Any:
+ """
+ Return ``rendered`` with every list that pydantic rendered from a set sorted.
+
+ ``rendered`` is pydantic's JSON rendering of ``value`` and is otherwise kept as
+ is, so pydantic stays the only renderer: bytes, dates, dict keys, ``NaN`` and
+ custom serializers come out exactly as a json-mode dump renders them. The one
+ thing pydantic cannot do stably is order a set. It lists the members in
+ iteration order, which for strings follows the interpreter's hash seed, and
+ every task attempt is a fresh process, so a ``set[str]`` would hash differently
+ on each attempt and never replay. ``value`` is walked alongside ``rendered``
+ only to find those lists; a branch whose rendering does not line up with the
+ object, such as one with a custom serializer, is left exactly as rendered.
+ """
+ if isinstance(value, (set, frozenset)):
+ if not isinstance(rendered, list) or len(rendered) != len(value):
+ return rendered
+ members = [_order_sets(member, item) for member, item in zip(value, rendered)]
+ return sorted(members, key=lambda member: json.dumps(member, sort_keys=True))
+ if isinstance(value, Mapping):
+ if not isinstance(rendered, dict):
+ return rendered
+ if len(rendered) != len(value):
+ # Distinct keys that render alike, such as 1 and "1": the digest could no
+ # longer tell those payloads apart, so refuse rather than hash either one.
+ raise TypeError("dict keys collide once rendered as JSON")
Review Comment:
A length mismatch here is not always a collision: a field serializer that filters a dict renders fewer keys than the object holds. A tool that returns a model whose `@field_serializer("headers")` drops `authorization` lands in `ToolReturnPart.content`, this raises, and every later model step fingerprints `None` with a warning about colliding keys that don't collide. Main hashed that history, and the rendered dict is also what the model receives. The docstring above says a branch like this is left as rendered. Raising only when the keys really collide, `len(to_jsonable_python({k: None for k in value}, bytes_mode="base64")) < len(value)`, and otherwise returning `rendered` keeps the refusal for `{1: "a", "1": "b"}` and gives main's digest back for the serializer case.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/fingerprint.py:
##########
@@ -33,24 +33,43 @@
Fields that pydantic-ai regenerates on every attempt (message-level
``timestamp``/``run_id``/``conversation_id`` and part-level ``timestamp``)
-are excluded from the fingerprint. Requests that cannot be serialized to
-JSON fingerprint as ``None``, which degrades that step to unverified
-positional replay (the pre-fingerprint behavior) rather than disabling
-caching.
+are excluded from the fingerprint. Every payload is rendered by pydantic in
+JSON mode before hashing, so values that are not JSON types but render the same
+way on every attempt -- a ``datetime`` or ``Decimal`` tool argument, a dataclass
+in ``tool_choice``, bytes in a ``BinaryContent`` -- still produce a usable
+fingerprint, and the message history hashes exactly as it always has. The one
+correction applied on top is that set members are ordered by their JSON encoding,
+because a set of strings iterates in an order that follows the interpreter's hash
+seed and every task attempt is a fresh process (see ``_order_sets``).
+
+A request that cannot be rendered fingerprints as ``None``: that step is
+neither replayed nor cached, and re-runs live instead of replaying without
+verification. A lazily validated ``Iterable`` argument lands there deliberately,
+since hashing it would consume the input the tool has not read yet, as does a
+value pydantic cannot serialize at all. On the model path this is seldom
+confined to one step, because model settings, the tool definitions and the
+message history are carried into every later request, so durable execution stops
+contributing anything for the rest of the run. A tool call is fingerprinted from its name, arguments and
+call id alone, so it can only lose its own step.
Review Comment:
"It can only lose its own step" is the claim the rst paragraph moved away from last round. If the live re-run returns something different, the result is in the history the following model steps fingerprint, so they miss too. Something like "so it cannot make another step unfingerprintable, though a re-run that returns something different still invalidates the model steps after it" would match the docs.
##########
providers/common/ai/tests/unit/common/ai/durable/test_fingerprint.py:
##########
@@ -210,3 +237,339 @@ def test_arg_order_does_not_matter(self):
assert fingerprint_tool_call("t", {"a": 1, "b": 2}, "id1") == fingerprint_tool_call(
"t", {"b": 2, "a": 1}, "id1"
)
+
+
+class TestPydanticNativeValues:
+ """Values that are not JSON types but hash the same on every attempt must still fingerprint.
+
+ Tool arguments reach ``fingerprint_tool_call`` already coerced by pydantic, and
+ ``tool_choice`` accepts a dataclass while genuinely affecting the response, so it
+ cannot be stripped as transport-only. Since a step that cannot be fingerprinted is
+ no longer cached at all, refusing these values would stop an ordinary typed tool
+ from ever being cached.
+ """
+
+ def test_datetime_tool_argument_fingerprints(self):
+ when = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": when}, "id1") is not None
+
+ def test_decimal_tool_argument_fingerprints(self):
+ assert fingerprint_tool_call("t", {"amount": Decimal("10.5")}, "id1") is not None
+
+ def test_datetime_tool_argument_is_stable_and_distinguishing(self):
+ early = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ late = datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc)
+
+ assert fingerprint_tool_call("t", {"when": early}, "id1") == fingerprint_tool_call(
+ "t", {"when": early}, "id1"
+ )
+ assert fingerprint_tool_call("t", {"when": early}, "id1") != fingerprint_tool_call(
+ "t", {"when": late}, "id1"
+ )
+
+ def test_tool_choice_dataclass_fingerprints(self):
+ fp = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["my_tool"])},
+ ModelRequestParameters(),
+ )
+
+ assert fp is not None
+
+ def test_tool_choice_dataclass_still_affects_the_fingerprint(self):
+ one = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["a"])},
+ ModelRequestParameters(),
+ )
+ other = fingerprint_model_request(
+ "m",
+ make_messages(),
+ {"tool_choice": ToolOrOutput(function_tools=["b"])},
+ ModelRequestParameters(),
+ )
+
+ assert one is not None
+ assert one != other
+
+ def test_value_pydantic_cannot_serialize_still_returns_none(self):
+ """Normalization must not turn a genuinely unserializable value into a hash."""
+ assert fingerprint_tool_call("t", {"v": object()}, "id1") is None
+
+ def test_plain_payload_digest_is_unchanged_by_normalization(self):
+ """Fingerprints recorded before normalization must still match, so cached entries survive."""
+ payload = {"model": "m", "args": {"b": [1, True, None, "x", 2.5]}, "settings": None}
+ pre_normalization = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+ assert _digest(_render(payload)) == pre_normalization
+
+ def test_dict_keys_render_as_a_json_mode_dump_renders_them(self):
+ by_day = {datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}
+
+ assert _render({"series": by_day}) == {"series": {"2026-01-01": 1.5, "2026-01-02": 2.5}}
+ assert _render({1: "a", "b": 2}) == {"1": "a", "b": 2}
+
+ def test_keys_that_collide_once_rendered_are_refused(self):
+ """``1`` and ``"1"`` render alike, so hashing either payload could replay the other."""
+ assert fingerprint_tool_call("t", {"d": {1: "a", "1": "b"}}, "id1") is None
+
+ def test_bytes_that_are_not_utf8_render_as_base64(self):
+ """Decoding bytes as text would raise on an image and stop the tool from being cached."""
+ assert _render({"image": _PNG}) == {"image": base64.urlsafe_b64encode(_PNG).decode()}
+ assert fingerprint_tool_call("t", {"image": _PNG}, "id1") is not None
+
+
+def _json_mode_reference(model_identifier, messages, model_request_parameters):
+ """The fingerprint main computes: pydantic's json-mode dump, hashed as is.
+
+ For anything that is not a set, fingerprints must equal this, or entries stored
+ by an earlier version stop matching and the first retry after an upgrade re-runs
+ the whole agent.
+ """
+ dumped = ModelMessagesTypeAdapter.dump_python(messages, mode="json")
+ stripped = [
+ {
+ **{k: v for k, v in message.items() if k not in ("timestamp", "run_id", "conversation_id")},
+ "parts": [{k: v for k, v in part.items() if k != "timestamp"} for part in message["parts"]],
+ }
+ for message in dumped
+ ]
+ params = TypeAdapter(ModelRequestParameters).dump_python(model_request_parameters, mode="json")
+ payload = {"model": model_identifier, "messages": stripped, "settings": None, "params": params}
+ return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+
+
+def _with_tool_return(content):
+ return [
+ ModelRequest(parts=[UserPromptPart(content="go")]),
+ ModelResponse(parts=[ToolCallPart(tool_name="t", args={}, tool_call_id="c1")]),
+ ModelRequest(parts=[ToolReturnPart(tool_name="t", content=content, tool_call_id="c1")]),
+ ]
+
+
+class TestMessageHistoryMatchesJsonModeDump:
+ """The message history must hash exactly as pydantic's json-mode dump renders it.
+
+ Each case here is something a hand-written renderer gets wrong: raw bytes (not
+ valid UTF-8), dict keys that are not strings, ``NaN``, tuples, and fields whose
+ serializer only applies in JSON mode, such as ``InstructionPart.id``.
+ """
+
+ @pytest.mark.parametrize(
+ "messages",
+ [
+ pytest.param(
+ [
+ ModelRequest(
+ parts=[
+ UserPromptPart(content=["look", BinaryContent(data=_PNG, media_type="image/png")])
+ ]
+ )
+ ],
+ id="binary-content-in-prompt",
+ ),
+ pytest.param(_with_tool_return(_PNG), id="bytes-tool-return"),
+ pytest.param(
+ _with_tool_return({datetime.date(2026, 1, 1): 1.5, datetime.date(2026, 1, 2): 2.5}),
+ id="date-keyed-tool-return",
+ ),
+ pytest.param(_with_tool_return({1: "a", "b": 2}), id="mixed-key-tool-return"),
+ pytest.param(_with_tool_return({"x": math.nan}), id="nan-tool-return"),
+ pytest.param(
+ _with_tool_return({"when": datetime.datetime(2026, 1, 1)}), id="datetime-tool-return"
+ ),
+ ],
+ )
+ def test_fingerprint_equals_the_json_mode_digest(self, messages):
+ fp = fingerprint_model_request("m", messages, None, ModelRequestParameters())
+
+ assert fp is not None
+ assert fp == _json_mode_reference("m", messages, ModelRequestParameters())
+
+ def test_request_parameters_hash_as_their_json_mode_dump(self):
+ """An agent's instructions reach the request parameters with a json-only serializer."""
+ seen = {}
+
+ class Spy(FunctionModel):
+ async def request(self, messages, model_settings, model_request_parameters):
+ seen.setdefault("messages", messages)
+ seen.setdefault("params", model_request_parameters)
+ return await super().request(messages, model_settings, model_request_parameters)
+
+ async def respond(messages, info):
+ return ModelResponse(parts=[TextPart(content="ok")])
+
+ Agent(Spy(respond), instructions="Be terse.").run_sync("hi")
+
+ fp = fingerprint_model_request("m", seen["messages"], None, seen["params"])
+
+ assert fp == _json_mode_reference("m", seen["messages"], seen["params"])
+
+ def test_tuple_parts_still_drop_part_timestamps(self):
+ """Message history passed as objects can hold its parts in a tuple."""
+ t1 = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+ t2 = datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc)
+
+ def history(timestamp):
+ return [ModelRequest(parts=(UserPromptPart(content="hi", timestamp=timestamp),))]
+
+ assert fingerprint_model_request(
+ "m", history(t1), None, ModelRequestParameters()
+ ) == fingerprint_model_request("m", history(t2), None, ModelRequestParameters())
+
+ def test_set_in_a_tool_return_hashes_as_its_ordered_members(self):
+ assert fingerprint_model_request(
+ "m", _with_tool_return({"tags": {"b", "c", "a"}}), None, ModelRequestParameters()
+ ) == fingerprint_model_request(
+ "m", _with_tool_return({"tags": ["a", "b", "c"]}), None, ModelRequestParameters()
+ )
+
+
+class TestSetMemberOrdering:
+ """Sets must hash in a fixed order rather than the interpreter's iteration order.
+
+ A ``set[str]`` iterates in an order derived from the process hash seed, and every
+ task attempt runs in a fresh process. Hashing that order would produce a digest
+ the next attempt never reproduces, so the step would re-run live on every retry
+ -- worse than declining to cache it, which at least costs nothing extra.
+ """
+
+ def test_set_hashes_as_its_ordered_members(self):
+ assert _render({"tags": {"beta", "alpha"}}) == {"tags": ["alpha", "beta"]}
+
+ def test_set_matches_the_equivalent_list(self):
+ assert _render({"tags": {"alpha", "beta", "gamma"}}) == _render({"tags": ["alpha", "beta", "gamma"]})
+
+ def test_frozenset_matches_set(self):
+ assert _render({"tags": frozenset({"a", "b"})}) == _render({"tags": {"b", "a"}})
+
+ def test_different_members_still_produce_different_digests(self):
+ assert _render({"tags": {"a", "b"}}) != _render({"tags": {"a", "c"}})
+
+ def test_set_nested_inside_a_list(self):
+ assert _render({"filters": [{"z", "y"}]}) == {"filters": [["y", "z"]]}
+
+ def test_set_inside_a_dataclass_field(self):
+ @dataclasses.dataclass
+ class Filter:
+ tags: set
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_set_inside_a_basemodel_field(self):
+ class Filter(pydantic.BaseModel):
+ tags: set[str]
+
+ assert _render(Filter(tags={"b", "a"})) == {"tags": ["a", "b"]}
+
+ def test_digest_is_stable_across_process_hash_seeds(self):
+ """The real proof: two fresh interpreters must agree, as two attempts would.
+
+ In-process comparisons cannot catch a hash-seed dependency, since one process
+ has one seed. The subprocess loads the module by path so it does not pay for
+ importing Airflow.
+ """
+ snippet = (
+ "import importlib.util;"
+ f"spec = importlib.util.spec_from_file_location('fp', r'{fingerprint_module.__file__}');"
+ "mod = importlib.util.module_from_spec(spec);"
+ "spec.loader.exec_module(mod);"
+ "print(mod._digest(mod._render({'tags': {'alpha', 'beta', 'gamma', 'delta'}})))"
+ )
+ digests = set()
+ for seed in ("0", "1", "2", "42"):
+ completed = subprocess.run(
+ [sys.executable, "-c", snippet],
+ capture_output=True,
+ text=True,
+ env={**os.environ, "PYTHONHASHSEED": seed, "PYTHONWARNINGS": "ignore"},
+ check=True,
+ )
+ digests.add(completed.stdout.strip().splitlines()[-1])
+
+ assert len(digests) == 1, f"digest depends on the hash seed: {digests}"
+
+ def test_set_returned_by_a_tool_is_stable_across_process_hash_seeds(self):
+ """The same proof through the message history, where a tool's set return lands."""
+ snippet = (
+ "import importlib.util;"
+ "from pydantic_ai.messages import ModelRequest, ToolReturnPart;"
+ "from pydantic_ai.models import ModelRequestParameters;"
+ f"spec = importlib.util.spec_from_file_location('fp', r'{fingerprint_module.__file__}');"
+ "mod = importlib.util.module_from_spec(spec);"
+ "spec.loader.exec_module(mod);"
+ "part = ToolReturnPart(tool_name='t', content={'alpha', 'beta', 'gamma', 'delta'}, tool_call_id='c1');"
+ "print(mod.fingerprint_model_request('m', [ModelRequest(parts=[part])], None, ModelRequestParameters()))"
+ )
+ digests = set()
+ for seed in ("0", "1", "2", "42"):
+ completed = subprocess.run(
+ [sys.executable, "-c", snippet],
+ capture_output=True,
+ text=True,
+ env={**os.environ, "PYTHONHASHSEED": seed, "PYTHONWARNINGS": "ignore"},
+ check=True,
+ )
+ digests.add(completed.stdout.strip().splitlines()[-1])
+
+ assert "None" not in digests
+ assert len(digests) == 1, f"digest depends on the hash seed: {digests}"
+
+
+class TestLazilyValidatedIterable:
+ """A lazily validated ``Iterable`` must be refused, not consumed.
+
+ ``to_jsonable_python`` drains an iterator to render it. Tool arguments are
+ fingerprinted before the tool runs, so draining one would hand the tool an
+ exhausted iterator and cache that wrong result under the fingerprint of the
+ full input. Declining to fingerprint leaves the step uncached instead.
+ """
+
+ def test_validator_iterator_argument_is_not_fingerprinted(self):
Review Comment:
Every case in this class passes the iterator at the top level of the args. Deleting the list/tuple, `BaseModel` or dataclass branch of `_refuse_iterators`, or the `_refuse_iterators(messages)` call, leaves the durable suite green, and each deletion brings back the drained-input bug for a nested parameter such as `list[Iterable[int]]` or a model field `ids: Iterable[int]`. Parametrizing over those shapes (plus the extras case) and adding a `fingerprint_model_request` case with a generator in `ToolReturnPart.content` would pin them. The generator case is also a fix worth recording: on main the model received `[]` there. Removing the `_order_sets` wrap on the request parameters is likewise not caught, even though `revealed_tool_names` is a `set[str]` in real requests. Putting a few names there in the subprocess seed test would guard it.
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -102,6 +102,33 @@ cache:
never replays responses that belong to a different conversation.
4. After successful completion, the cached steps are deleted.
+Fingerprints are computed from pydantic's JSON rendering of each value, the same
+rendering a json-mode dump produces, so ordinary types that are not JSON -- a
+``datetime`` or ``Decimal`` tool argument, a dataclass in ``tool_choice``, the
+bytes in a ``BinaryContent``, a dict keyed by date -- fingerprint normally, and
+entries cached by an earlier version still match. The one adjustment is that the
+members of a ``set`` are ordered before hashing, so a set matches on a later
+attempt too. If a value cannot be rendered at all, that step is not cached, and on
+retry it runs live rather than replaying an unverified entry. A parameter annotated ``Iterable[...]`` is one such case:
+pydantic validates it lazily, and reading it in order to hash it would consume
+the input the tool itself has not read yet, so the step runs live instead of
+being cached.
+
+On the model path this is rarely confined to a single step: the causes are such a
+value in ``model_settings``, which is attached to every request, in the tool
+definitions the request carries, or in the message history, which every later
+request carries forward. Any one of them degrades all subsequent model steps the
+same way, leaving durable execution with nothing to
+replay, so the retry re-runs the agent at full cost. The
Review Comment:
"Nothing to replay" and "re-runs the agent at full cost" hold when the cause is in `model_settings` or the tool definitions, since those are there from the first request. When it enters the history mid-run, from a tool return for example, the steps before it still replay and only the steps after it run live. Something like "degrades every model step from that point on, so the retry re-runs the agent from there at full cost, and from the start when the cause is in the settings or tool definitions" would cover both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org