Lee-W commented on code in PR #76863:
URL: https://github.com/apache/airflow/pull/76863#discussion_r4533297455
##########
providers/common/ai/src/airflow/providers/common/ai/durable/journal.py:
##########
@@ -279,6 +284,9 @@ def __init__(self, journal: DurableJournal, key: str) ->
None:
# The furthest position this attempt recorded a step at, and every
position it recorded.
self._last_saved = -1
self._saved: set[int] = set()
+ # Every position this attempt sent a write for, including writes
reported as failed,
+ # which may still have landed; cleanup of a run no earlier attempt
started deletes these.
+ self._written: set[int] = set()
Review Comment:
```suggestion
self._written_positions: set[int] = set()
```
nit: `_written` took me some time to understand what this actaully is
##########
providers/common/ai/src/airflow/providers/common/ai/durable/journal.py:
##########
@@ -557,14 +573,20 @@ def _count_tool_failure(self, step: JournalStep | None)
-> _ToolFailure | None:
def cleanup(self) -> None:
"""Delete this run's steps. Call only once the run, and anything that
consumes it, has succeeded."""
+ self.journal._delete_runs([self])
+
+ def _keys(self) -> list[str]:
Review Comment:
```suggestion
def _get_keys(self) -> list[str]:
```
##########
providers/common/ai/docs/durable_execution.rst:
##########
@@ -100,6 +100,16 @@ On **Airflow 3.3 and later** the journal is kept in the
task instance. No configuration is needed. Each step is written to the Airflow
metadata
database; for agents with large model responses or tool results, set
``[workers] state_store_backend`` to offload the values to external storage
instead.
+Without it, a step larger than ``[state_store] max_value_storage_bytes`` (64
KB by
+default) is still written to the database, and the task log warns that it
exceeds the
+limit.
+
+Every step the agent completes is one write to the task state store, a call
the task
+makes to the API server while the agent waits. A first attempt reads back only
the list
+of agent runs the task started (and, in ``AgentOperator``, the run id), since
no earlier
+attempt recorded anything; a retry also looks up each step of a run an earlier
attempt
+started, and stops looking once the run takes a different path from that
attempt.
+When the task succeeds, its journal is deleted, one call per entry.
Review Comment:
Nit: "one call per entry" holds on >= 3.3 only. < 3.3 the journal file is
rewritten once, as the PR description says.
The paragraph is inside the “3.3 and later” section but mixes statements
true for both backends with 3.3-only ones. Move the first-attempt sentence
ahead of the 3.3-only sentences and scope the delete sentence, e.g. “When the
task succeeds, its journal is deleted: on 3.3 and later that is one request per
entry; before 3.3 the journal file is rewritten once.”
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
