This is an automated email from the ASF dual-hosted git repository. kaxil pushed a commit to branch anthropic-interrupt-before-archive in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ca5d7c34401447333951643a6abf051d8b1b0b73 Author: Kaxil Naik <[email protected]> AuthorDate: Wed Aug 12 01:13:35 2026 +0530 Interrupt a running Anthropic session before archiving it The API refuses to archive **or** delete a session while its status is ``running``, rejecting both with a 400. ``AnthropicAgentSessionOperator`` treats archiving as best-effort teardown on every failure path, so a session that will not stop on its own is left behind with no way to release it -- and it keeps accruing ``active_seconds``, which the SDK describes as the duration the session's runtime cost is priced on. Observed directly: two sessions halted by a $0.01 budget sat at status ``running`` for over 40 minutes with model spend frozen at the ceiling while ``active_seconds`` kept climbing. ``archive`` and ``delete`` both returned 400. Sending ``user.interrupt`` moved them to ``idle`` immediately, after which archiving succeeded. ``AnthropicHook.archive_session`` now interrupts and retries when the first archive attempt fails, and ``interrupt_session`` exposes the event on its own. The retry is bounded, so an unarchivable session still surfaces its error rather than looping. This is reachable well beyond budgets -- any session still working when a task fails, times out, or is killed hits the same 400 -- but a budget halt makes it routine, because that is a session which has stopped spending without stopping. The interrupt-then-archive sequence was verified by hand against the live API on two genuinely stuck sessions. Its wiring through ``archive_session`` is covered by unit tests rather than a live run. --- .../airflow/providers/anthropic/hooks/anthropic.py | 36 +++++++++++++++++++++- .../tests/unit/anthropic/hooks/test_anthropic.py | 30 ++++++++++++++++++ 2 files changed, 65 insertions(+), 1 deletion(-) diff --git a/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py b/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py index dac9549fd45..2992ace4907 100644 --- a/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py +++ b/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py @@ -710,15 +710,49 @@ class AnthropicHook(BaseHook): session_id, events=cast("list[BetaManagedAgentsEventParams]", [event]) ) + def interrupt_session(self, session_id: str) -> Any: + """ + Send ``user.interrupt`` to pause a running session. + + The API refuses to archive or delete a session while it is ``running``, so this is + the only way to release one that is not going to stop on its own -- see + :meth:`archive_session`. + """ + self._require_first_party("Managed Agents") + return self.send_event(session_id, {"type": "user.interrupt"}) + def archive_session(self, session_id: str) -> BetaManagedAgentsSession: """ Archive a session (frees the server-side container). Best-effort teardown. Returns the archived session, which carries its final ``usage`` -- so a caller tearing a session down does not need a separate retrieve to report what it spent. + + A ``running`` session cannot be archived (nor deleted): the API rejects both with a + 400. When that happens this interrupts the session and retries once, because a + session that will not stop on its own otherwise accrues billable runtime with no way + to release it. """ self._require_first_party("Managed Agents") - return self._first_party_conn.beta.sessions.archive(session_id) + try: + return self._first_party_conn.beta.sessions.archive(session_id) + except Exception as e: + self.log.info("Archiving session %s failed (%s); interrupting and retrying.", session_id, e) + self.interrupt_session(session_id) + return self._wait_for_archive(session_id) + + def _wait_for_archive( + self, session_id: str, attempts: int = 6, wait_seconds: float = 5 + ) -> BetaManagedAgentsSession: + """Retry archiving while the interrupt takes effect; the status change is not instant.""" + for attempt in range(attempts): + try: + return self._first_party_conn.beta.sessions.archive(session_id) + except Exception: + if attempt == attempts - 1: + raise + time.sleep(wait_seconds) + raise AnthropicError(f"Could not archive session {session_id}.") # pragma: no cover def _latest_idle_reason(self, session_id: str, kickoff_event_id: str | None) -> str | None: """ diff --git a/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py b/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py index 5442aaeff40..2f3ca465b5a 100644 --- a/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py +++ b/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py @@ -266,6 +266,36 @@ class TestBuildBudget: build_budget(amount) +class TestArchiveSession: + def test_archives_directly_when_the_session_is_stoppable(self): + hook, client = _make_hook() + hook.archive_session("sess_1") + client.beta.sessions.archive.assert_called_once_with("sess_1") + client.beta.sessions.events.send.assert_not_called() + + @mock.patch(f"{HOOK_PATH}.time.sleep", autospec=True) + def test_interrupts_and_retries_when_the_session_is_running(self, mock_sleep): + # A running session is refused by both archive and delete, so without the interrupt + # it keeps accruing billable runtime with no way to release it. + hook, client = _make_hook() + archived = object() + client.beta.sessions.archive.side_effect = [RuntimeError("400 running"), archived] + assert hook.archive_session("sess_1") is archived + client.beta.sessions.events.send.assert_called_once_with( + "sess_1", events=[{"type": "user.interrupt"}] + ) + assert client.beta.sessions.archive.call_count == 2 + + @mock.patch(f"{HOOK_PATH}.time.sleep", autospec=True) + def test_gives_up_after_the_retry_budget(self, mock_sleep): + hook, client = _make_hook() + client.beta.sessions.archive.side_effect = RuntimeError("400 running") + with pytest.raises(RuntimeError, match="400 running"): + hook.archive_session("sess_1") + # one initial attempt plus the bounded retry loop + assert client.beta.sessions.archive.call_count == 7 + + class TestUpdateSession: def test_only_passes_supplied_keys(self): # The API distinguishes omitted (preserve) from None (clear), so an unmentioned
