amoghrajesh commented on code in PR #76786:
URL: https://github.com/apache/airflow/pull/76786#discussion_r4533979409
##########
providers/common/ai/src/airflow/providers/common/ai/durable/openai_agents.py:
##########
@@ -216,11 +240,182 @@ def run_durably_sync(
return run_coroutine_sync(run_durably(starting_agent, input,
run_config=run_config, **kwargs))
+def _answer_approvals(
+ result: RunResult, durable_run: DurableRun, answered: set[str], *,
has_session: bool
+) -> RunState[Any]:
+ """
+ Return the run's state with every pending tool call approved or rejected
by a person.
+
+ A call is asked through ``request_input()``, which pauses the task the
first time and
+ returns the answer once the task resumes; its key goes into ``answered``.
A call whose
+ result is already recorded was approved and ran in an earlier run of the
task, so it is
+ approved again without asking.
+
+ :raises UnsupportedToolDeferralError: When the run cannot pause for the
approval.
+ :raises DurableJournalError: When the task resumed with an answer but the
run did not come
+ back to the call it answers.
+ """
+ if (request_input := _get_request_input()) is None:
+ _cannot_pause(
+ result,
+ durable_run,
+ "pausing a task for input needs Airflow 3.4+ and a standard
provider that has request_input()",
+ )
+ if (reason := _why_approval_cannot_pause(result, has_session)) is not None:
+ _cannot_pause(result, durable_run, reason)
+ durable_run.pause()
+ state = result.to_state()
+ for item in result.interruptions:
+ call = item.raw_item
+ if durable_run.has_recorded(call.call_id):
+ log.info(
+ "Durable: %s ran after it was approved in an earlier run of
this task; approving it again",
+ item.tool_name,
+ )
+ state.approve(item)
+ continue
+ # The call comes back the same on every resume, as the answer that
made it replays.
+ # Its fingerprint keeps an answer from applying to a different call
that reuses its id.
+ key =
f"{durable_run.key}:{call.call_id}:{_build_tool_fingerprint(call.name,
call.arguments, call.call_id)}"
+ try:
+ answer = request_input(
+ subject=f"{item.agent.name} wants to call {item.tool_name}",
+ body=_describe_call(item),
+ options=[LLMApprovalMixin.APPROVE, LLMApprovalMixin.REJECT],
+ # An empty default, not None: some UI versions show a null one
as "[object Object]".
+ params={
+ "reason": Param(
+ "",
+ type=["string", "null"],
+ description="Sent to the agent when you reject the
call (optional).",
+ )
+ },
+ key=key,
+ )
+ except RuntimeError as error:
Review Comment:
This is meant to catch only `AnswerLog.check_latest_read()`, but it catches
every `RuntimeError` that `request_input()` raises. Any other one becomes a
`DurableJournalError`, which is an `AirflowFailException`, so the task fails
with no retry and a wrong "did not come back" message. Check
`answer_log.latest` yourself before calling `request_input()`, the way
`_answer_left_unread` already does, instead of catching a broad exception type.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/openai_agents.py:
##########
@@ -216,11 +240,182 @@ def run_durably_sync(
return run_coroutine_sync(run_durably(starting_agent, input,
run_config=run_config, **kwargs))
+def _answer_approvals(
+ result: RunResult, durable_run: DurableRun, answered: set[str], *,
has_session: bool
+) -> RunState[Any]:
+ """
+ Return the run's state with every pending tool call approved or rejected
by a person.
+
+ A call is asked through ``request_input()``, which pauses the task the
first time and
+ returns the answer once the task resumes; its key goes into ``answered``.
A call whose
+ result is already recorded was approved and ran in an earlier run of the
task, so it is
+ approved again without asking.
+
+ :raises UnsupportedToolDeferralError: When the run cannot pause for the
approval.
+ :raises DurableJournalError: When the task resumed with an answer but the
run did not come
+ back to the call it answers.
+ """
+ if (request_input := _get_request_input()) is None:
+ _cannot_pause(
+ result,
+ durable_run,
+ "pausing a task for input needs Airflow 3.4+ and a standard
provider that has request_input()",
+ )
+ if (reason := _why_approval_cannot_pause(result, has_session)) is not None:
+ _cannot_pause(result, durable_run, reason)
+ durable_run.pause()
+ state = result.to_state()
+ for item in result.interruptions:
+ call = item.raw_item
+ if durable_run.has_recorded(call.call_id):
+ log.info(
+ "Durable: %s ran after it was approved in an earlier run of
this task; approving it again",
+ item.tool_name,
+ )
+ state.approve(item)
+ continue
+ # The call comes back the same on every resume, as the answer that
made it replays.
+ # Its fingerprint keeps an answer from applying to a different call
that reuses its id.
+ key =
f"{durable_run.key}:{call.call_id}:{_build_tool_fingerprint(call.name,
call.arguments, call.call_id)}"
+ try:
+ answer = request_input(
+ subject=f"{item.agent.name} wants to call {item.tool_name}",
+ body=_describe_call(item),
+ options=[LLMApprovalMixin.APPROVE, LLMApprovalMixin.REJECT],
+ # An empty default, not None: some UI versions show a null one
as "[object Object]".
+ params={
+ "reason": Param(
+ "",
+ type=["string", "null"],
+ description="Sent to the agent when you reject the
call (optional).",
+ )
+ },
+ key=key,
+ )
+ except RuntimeError as error:
+ # Raised by request_input() when the answer the task resumed with
is for a call that
+ # this run never reached; asking again would never end.
+ _did_not_come_back(durable_run, error)
+ answered.add(key)
+ responder = LLMApprovalMixin._describe_responder(dict(answer))
Review Comment:
I suggest `_describe_responder` be made public or locally defined to avoid
future breakage.
##########
providers/common/ai/src/airflow/providers/common/ai/durable/openai_agents.py:
##########
@@ -216,11 +240,182 @@ def run_durably_sync(
return run_coroutine_sync(run_durably(starting_agent, input,
run_config=run_config, **kwargs))
+def _answer_approvals(
+ result: RunResult, durable_run: DurableRun, answered: set[str], *,
has_session: bool
+) -> RunState[Any]:
+ """
+ Return the run's state with every pending tool call approved or rejected
by a person.
+
+ A call is asked through ``request_input()``, which pauses the task the
first time and
+ returns the answer once the task resumes; its key goes into ``answered``.
A call whose
+ result is already recorded was approved and ran in an earlier run of the
task, so it is
+ approved again without asking.
+
+ :raises UnsupportedToolDeferralError: When the run cannot pause for the
approval.
+ :raises DurableJournalError: When the task resumed with an answer but the
run did not come
+ back to the call it answers.
+ """
+ if (request_input := _get_request_input()) is None:
+ _cannot_pause(
+ result,
+ durable_run,
+ "pausing a task for input needs Airflow 3.4+ and a standard
provider that has request_input()",
+ )
+ if (reason := _why_approval_cannot_pause(result, has_session)) is not None:
+ _cannot_pause(result, durable_run, reason)
+ durable_run.pause()
+ state = result.to_state()
+ for item in result.interruptions:
+ call = item.raw_item
+ if durable_run.has_recorded(call.call_id):
+ log.info(
+ "Durable: %s ran after it was approved in an earlier run of
this task; approving it again",
+ item.tool_name,
+ )
+ state.approve(item)
+ continue
+ # The call comes back the same on every resume, as the answer that
made it replays.
+ # Its fingerprint keeps an answer from applying to a different call
that reuses its id.
+ key =
f"{durable_run.key}:{call.call_id}:{_build_tool_fingerprint(call.name,
call.arguments, call.call_id)}"
+ try:
+ answer = request_input(
+ subject=f"{item.agent.name} wants to call {item.tool_name}",
+ body=_describe_call(item),
+ options=[LLMApprovalMixin.APPROVE, LLMApprovalMixin.REJECT],
+ # An empty default, not None: some UI versions show a null one
as "[object Object]".
+ params={
+ "reason": Param(
+ "",
+ type=["string", "null"],
+ description="Sent to the agent when you reject the
call (optional).",
+ )
+ },
+ key=key,
+ )
+ except RuntimeError as error:
+ # Raised by request_input() when the answer the task resumed with
is for a call that
+ # this run never reached; asking again would never end.
+ _did_not_come_back(durable_run, error)
+ answered.add(key)
+ responder = LLMApprovalMixin._describe_responder(dict(answer))
+ if answer["chosen_options"] == [LLMApprovalMixin.APPROVE]:
+ log.info("Durable: %s approved by %s", item.tool_name, responder)
+ state.approve(item)
+ else:
+ log.info("Durable: %s rejected by %s", item.tool_name, responder)
+ # A field left empty comes back as None or "", and some UI
versions send the param's spec.
+ reason = answer["params_input"].get("reason")
+ reason = reason.strip() if isinstance(reason, str) else ""
+ state.reject(item, rejection_message=reason or None)
+ return state
+
+
+def _answer_left_unread(durable_run: DurableRun, answered: set[str]) -> bool:
+ """Whether the task resumed with an answer to an approval of this run that
the run never asked again for."""
+ try:
+ # Optional: the answer log comes with request_input(), newer than the
standard provider common.ai requires.
+ from airflow.providers.standard.utils.hitl import answer_log_var
+ except ImportError:
+ return False
+ answer_log = answer_log_var.get()
+ return (
+ answer_log is not None
+ and answer_log.latest.startswith(f"{durable_run.key}:")
+ and answer_log.latest not in answered
+ )
+
+
+def _did_not_come_back(durable_run: DurableRun, cause: BaseException | None =
None) -> NoReturn:
+ """
+ Fail the task resumed with an answer whose call the run never reached
again.
+
+ ``cause`` is set when the run stopped at another call that needs approval
(a clear asks
+ about that one), and ``None`` when it finished without one (a clear
replays the result).
+ """
+ then = (
+ "Clear the task to ask about the call it stopped at instead."
+ if cause is not None
+ else "It finished without that call; clearing the task replays that
result without asking."
+ )
+ error = DurableJournalError(
+ "The task resumed with an answer to a tool approval, but the agent run
did not come back "
+ "to the call it answers: a step before that call gave a different
result this time, such "
+ "as a tool that always runs live, a prompt built from live data, or a
change to the Dag "
+ f"while the task waited. {then}"
+ )
+ durable_run.fail(error)
+ raise error from cause
+
+
+def _get_request_input() -> Callable[..., InputResponse] | None:
+ """Return the standard provider's ``request_input()``, or ``None`` where
it cannot pause a task."""
+ if not AIRFLOW_V_3_4_PLUS:
+ return None
+ try:
+ # Optional: request_input() is newer than the standard provider
release common.ai requires.
+ from airflow.providers.standard.operators.hitl import request_input
+ except ImportError:
+ return None
+ return request_input
+
+
+def _cannot_pause(result: RunResult, durable_run: DurableRun, reason: str) ->
NoReturn:
+ tools = ", ".join(sorted({f"{item.tool_name} ({item.agent.name})" for item
in result.interruptions}))
+ error = UnsupportedToolDeferralError(
+ f"The agent stopped to ask for approval to call {tools}, and
run_durably() cannot "
+ f"pause the task for it: {reason}. Remove the approval requirement
(needs_approval, "
+ "or require_approval on an MCP server), or call Runner.run and handle "
+ "result.interruptions yourself."
+ )
+ # Counted against the model's last answer, so a clear asks the model again
from there:
+ # what was recorded after it, such as an agent tool's answer given while
it waited for
+ # approval, never replays, and the turns before it still do.
+ durable_run.fail(error, model_at_fault=True)
+ raise error
+
+
+def _why_approval_cannot_pause(result: RunResult, has_session: bool) -> str |
None:
+ if has_session:
+ return (
+ "a run with a session asks the model again on every resume, so the
call it paused on "
+ "never comes back to receive the answer"
+ )
+ for item in result.interruptions:
+ if not isinstance(item.raw_item, ResponseFunctionToolCall) or not
getattr(
+ item.agent, _DURABLE_MARKER, False
+ ):
+ return (
+ f"{item.tool_name} is not a function tool of an agent whose
model run_durably() "
+ "records, such as an agent used through as_tool() or reached
through handoff(), "
+ "so its call would not come back the same when the task
resumes"
+ )
+ task = get_current_context()["task"]
Review Comment:
`get_current_context()` raises a raw `RuntimeError` when a journal is set
with `journal_scope` outside a task, for example, in a user's own unit test on
3.4+. Before this PR, the same case gave a clear
`UnsupportedToolDeferralError`. Catch it and return a reason, so
`_cannot_pause` runs and calls `durable_run.fail`.
--
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]
