feat(temporal): opt-in continue-as-new for long-lived agent workflows

Long-lived chat/session agents run as a single Temporal workflow that stays
open indefinitely, so their event history grows until it hits Temporal's
~50k-event / 50MB limit and the workflow stalls. This adds an opt-in
continue-as-new pattern on BaseWorkflow that recycles the history so a session
can stay open forever.

It is opt-in by design: an agent gets recycling only by calling
run_until_complete from its @workflow.run instead of a bare
wait_condition(timeout=None). There is no env flag.

BaseWorkflow helpers:
- run_until_complete(*args, is_complete, can_recycle=True, timeout=None): keep the
  workflow open; recycle history when Temporal suggests it. timeout mirrors the old
  wait_condition(timeout=...); can_recycle lets an agent opt out when a recycle
  prerequisite is missing.
- should_continue_as_new(): recycle when workflow.info().is_continue_as_new_suggested().
- drain_and_continue_as_new(): waits all_handlers_finished (so an in-flight turn is
  not lost) and re-checks completion before workflow.continue_as_new.
- recycling_active(): workflow.patched() gate, so a workflow that started before an
  agent adopted run_until_complete replays its original command stream unchanged.
- is_continued_run(): the hook agents use to gate state rehydration after a recycle.

Restoring state after a recycle is framework-specific (rebuild from adk.messages,
an adk.state snapshot, or a framework's own memory, e.g. a LangGraph checkpointer)
and is intentionally left to follow-up PRs, one per integration. The 000_hello_acp
example adopts the pattern; it keeps no cross-turn state, so it needs no rehydration.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: danielmillerp

examples/tutorials/10_async/10_temporal/000_hello_acp/project/workflow.py

@@ -62,10 +62,13 @@ async def on_task_create(self, params: CreateTaskParams) -> None:
             ),
         )
 
-        # 2. Wait for the task to be completed indefinitely. If we don't do this the workflow will close as soon as this function returns. Temporal can run hundreds of millions of workflows in parallel, so you don't need to worry about too many workflows running at once.
-
-        # Thus, if you want this agent to field events indefinitely (or for a long time) you need to wait for a condition to be met.
-        await workflow.wait_condition(
-            lambda: self._complete_task,
-            timeout=None, # Set a timeout if you want to prevent the task from running indefinitely. Generally this is not needed. Temporal can run hundreds of millions of workflows in parallel and more. Only do this if you have a specific reason to do so.
-        )
+        # 2. Keep the workflow open to field events. We use run_until_complete
+        # instead of a bare wait_condition: it still waits indefinitely, but also
+        # recycles the Temporal event history via continue-as-new before it hits the
+        # ~50k-event / 50MB limit, so this chat can stay open forever. Adopting
+        # run_until_complete IS the opt-in — agents that keep the old wait_condition
+        # never recycle. This agent keeps no cross-turn state, so nothing needs
+        # restoring across a recycle and `params` is the only carry-forward. (Agents
+        # that DO keep state rebuild it at the top of @workflow.run, gated on
+        # self.is_continued_run() — framework-specific, handled per-agent.)
+        await self.run_until_complete(params, is_complete=lambda: self._complete_task)

src/agentex/lib/core/temporal/workflows/workflow.py

@@ -1,4 +1,8 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
+from typing import Any, Callable
+from datetime import timedelta
 
 from temporalio import workflow
 
@@ -8,6 +12,15 @@
 
 logger = make_logger(__name__)
 
+# Patch identifier gating the continue-as-new recycle path. Workflows that were
+# already running before this code shipped have no record of this patch in their
+# event history; `workflow.patched()` lets the new drain + continue_as_new branch
+# be introduced without breaking determinism when those in-flight executions
+# replay against the new code. Gate the decision ONCE in the agent's @workflow.run
+# (see the 010_agent_chat reference agent) so pre-patch runs keep their old
+# behaviour and never spin.
+CONTINUE_AS_NEW_PATCH_ID = "agentex-base-workflow-continue-as-new-v1"
+
 
 class BaseWorkflow(ABC):
     def __init__(
@@ -24,3 +37,162 @@ async def on_task_event_send(self, params: SendEventParams) -> None:
     @abstractmethod
     async def on_task_create(self, params: CreateTaskParams) -> None:
         raise NotImplementedError
+
+    # ------------------------------------------------------------------ #
+    # Continue-as-new lifecycle helpers                                   #
+    #                                                                     #
+    # These let a long-lived chat/session workflow recycle its event      #
+    # history so it can stay open indefinitely without hitting Temporal's #
+    # ~50k-event / 50MB history limit. They are OPT-IN: an agent gets     #
+    # recycling only by calling `run_until_complete` from its             #
+    # `@workflow.run` instead of the usual indefinite `wait_condition`.   #
+    # The SDK owns the hard Temporal mechanics (recycle decision,         #
+    # draining in-flight handlers, the continue_as_new call, patch        #
+    # gating). Restoring state after a recycle is the AGENT's job and is  #
+    # framework-specific (rebuild from `adk.messages`, an `adk.state`     #
+    # snapshot, or a framework's own memory) — see `is_continued_run`.    #
+    # The 000_hello_acp example shows the minimal stateless adoption.     #
+    # ------------------------------------------------------------------ #
+
+    def should_continue_as_new(self) -> bool:
+        """Whether this run should recycle its event history via continue-as-new.
+
+        True when Temporal suggests it: ``is_continue_as_new_suggested()`` fires as
+        the event history approaches the server's size/count limit, so we let
+        Temporal own the threshold rather than configuring one ourselves.
+
+        This reads only a deterministic ``workflow.info()`` value and emits no
+        commands, so it is safe to use directly as a ``workflow.wait_condition``
+        predicate, e.g.::
+
+            await workflow.wait_condition(
+                lambda: self._complete_task or self.should_continue_as_new()
+            )
+        """
+        return workflow.info().is_continue_as_new_suggested()
+
+    async def drain_and_continue_as_new(
+        self,
+        *args: Any,
+        is_complete: Callable[[], bool] | None = None,
+    ) -> None:
+        """Drain in-flight signal handlers, then continue-as-new.
+
+        Call this from the agent's ``@workflow.run`` once the run loop wakes for a
+        recycle (see :meth:`should_continue_as_new`). ``args`` are forwarded
+        verbatim to ``workflow.continue_as_new`` and become the new run's input, so
+        pass whatever your ``@workflow.run`` signature expects — typically the
+        original ``CreateTaskParams`` (the new run keeps the same workflow id / task
+        id and re-hydrates its state from ``adk.state``).
+
+        IMPORTANT: keep your data OUTSIDE workflow state BEFORE calling this —
+        messages in ``adk.messages`` and any other state in ``adk.state``.
+        In-workflow attributes do NOT survive the recycle; only the forwarded
+        ``args`` do.
+
+        Waits on ``all_handlers_finished`` first so an in-flight turn (a signal
+        handler still running an activity) is never lost or duplicated across the
+        recycle boundary. ``workflow.continue_as_new`` raises to end the run, so
+        this never returns normally — EXCEPT when ``is_complete`` is given and
+        returns True after draining: a completion signal can arrive while we wait
+        for the drain, and the recycled run would start fresh (losing that
+        completion), so in that case we return without recycling and let the caller
+        finish.
+        """
+        # Don't recycle until any signal handler still running has finished, so a
+        # message mid-flight at the boundary is carried into the next run intact.
+        await workflow.wait_condition(workflow.all_handlers_finished)
+        # A completion signal may have landed during the drain — re-check before
+        # recycling so a workflow that should finish isn't kept open by the recycle.
+        if is_complete is not None and is_complete():
+            return
+        logger.info(
+            "Recycling workflow via continue-as-new "
+            f"(history_length={workflow.info().get_current_history_length()}, "
+            f"run_id={workflow.info().run_id})"
+        )
+        workflow.continue_as_new(*args)
+
+    async def run_until_complete(
+        self,
+        *continue_as_new_args: Any,
+        is_complete: Callable[[], bool],
+        can_recycle: bool = True,
+        timeout: timedelta | None = None,
+    ) -> None:
+        """Keep the workflow open to field events, recycling history as needed.
+
+        Drop-in replacement for the usual ``await workflow.wait_condition(
+        lambda: self._complete_task, timeout=None)`` at the end of an agent's
+        ``@workflow.run``. ``is_complete`` is a no-arg predicate (typically
+        ``lambda: self._complete_task``); ``continue_as_new_args`` are forwarded to
+        continue-as-new on recycle (typically the original ``CreateTaskParams``).
+
+        Adopting this method IS the opt-in to recycling — there is no separate
+        flag. An agent that keeps the old indefinite ``wait_condition`` never
+        recycles. The recycle path is gated behind ``workflow.patched(...)`` so a
+        workflow that started before the agent adopted this keeps waiting the old
+        way and never hits a non-determinism error on replay.
+
+        ``can_recycle`` lets an agent declare that recycling is unsafe right now and
+        fall back to the plain wait. Use it when a prerequisite for surviving a
+        recycle is missing — e.g. an agent that keeps non-message state in
+        ``adk.state`` (keyed by task + agent) has nothing to persist into without an
+        ``AGENT_ID``, so it should pass ``can_recycle=bool(environment_variables.AGENT_ID)``
+        rather than recycle and silently drop that state on the first
+        continue-as-new.
+
+        ``timeout`` optionally bounds how long the workflow waits with no progress
+        (mirrors the old ``wait_condition(timeout=...)``). Leave it None to wait
+        indefinitely — generally what you want, Temporal can keep huge numbers of
+        idle workflows open. Set it only if you have a reason to cap the task; on
+        expiry ``wait_condition`` raises ``asyncio.TimeoutError`` like before.
+
+        Persist anything you need across a recycle OUTSIDE workflow state first —
+        messages in ``adk.messages``, other state in ``adk.state`` — and rebuild it
+        at the top of ``@workflow.run``.
+        """
+        if not can_recycle or not self.recycling_active():
+            await workflow.wait_condition(is_complete, timeout=timeout)
+            return
+        while True:
+            await workflow.wait_condition(
+                lambda: is_complete() or self.should_continue_as_new(),
+                timeout=timeout,
+            )
+            if is_complete():
+                return
+            # Drains in-flight handlers, then continue-as-new (raises; never
+            # returns) — UNLESS a completion signal arrived during the drain, in
+            # which case it returns here and the next loop iteration completes.
+            await self.drain_and_continue_as_new(
+                *continue_as_new_args, is_complete=is_complete
+            )
+            if is_complete():
+                return
+
+    def recycling_active(self) -> bool:
+        """Whether continue-as-new machinery should run for this run.
+
+        ``True`` only when this run is on patched code (``workflow.patched``). This
+        is the determinism gate: a workflow that started BEFORE an agent adopted
+        ``run_until_complete`` has no patch marker in its history, so this returns
+        ``False`` there and it replays its original command stream unchanged instead
+        of trying to recycle. Calling it is replay-safe — on a pre-patch run
+        ``workflow.patched`` returns False and emits no command.
+        """
+        return workflow.patched(CONTINUE_AS_NEW_PATCH_ID)
+
+    def is_continued_run(self) -> bool:
+        """Whether this run was produced by a continue-as-new from a prior run.
+
+        True only on a recycled run (``workflow.info().continued_run_id`` is set),
+        False on the original run a client created. This is the gate for any new,
+        activity-emitting prologue work in ``@workflow.run`` (re-hydrating state
+        from ``adk.state`` / ``adk.messages``): such work is only needed on a
+        continued run, and gating it here means an original run emits no new
+        commands. That keeps a pre-existing in-flight workflow (started before this
+        code shipped) deterministic on replay even with continue-as-new enabled,
+        since its original run skips the new prologue entirely.
+        """
+        return workflow.info().continued_run_id is not None

tests/lib/core/temporal/test_base_workflow_continue_as_new.py

@@ -0,0 +1,97 @@
+"""Unit tests for BaseWorkflow's continue-as-new lifecycle helpers.
+
+These exercise the pure decision helpers (``should_continue_as_new``,
+``recycling_active``, ``is_continued_run``) by faking ``workflow.info()`` /
+``workflow.patched()`` so we don't need a running Temporal server. The drain +
+``workflow.continue_as_new`` mechanics in ``drain_and_continue_as_new`` /
+``run_until_complete`` are best covered by a replay/integration test against a
+Temporal test environment (a follow-up).
+"""
+
+from __future__ import annotations
+
+from typing import override
+
+import pytest
+
+from agentex.lib.core.temporal.workflows import workflow as base_workflow_module
+from agentex.lib.core.temporal.workflows.workflow import BaseWorkflow
+
+
+class _ConcreteWorkflow(BaseWorkflow):
+    """Minimal concrete subclass so we can instantiate the ABC in a test."""
+
+    def __init__(self) -> None:
+        self.display_name = "test"
+
+    @override
+    async def on_task_event_send(self, params) -> None:  # pragma: no cover - unused
+        raise NotImplementedError
+
+    @override
+    async def on_task_create(self, params) -> None:  # pragma: no cover - unused
+        raise NotImplementedError
+
+
+class _FakeInfo:
+    def __init__(self, *, suggested: bool, continued_run_id: str | None = None) -> None:
+        self._suggested = suggested
+        self.continued_run_id = continued_run_id
+
+    def is_continue_as_new_suggested(self) -> bool:
+        return self._suggested
+
+
+@pytest.fixture
+def patch_info(monkeypatch):
+    """Patch ``workflow.info`` used inside the BaseWorkflow module."""
+
+    def _apply(*, suggested: bool, continued_run_id: str | None = None) -> None:
+        monkeypatch.setattr(
+            base_workflow_module.workflow,
+            "info",
+            lambda: _FakeInfo(suggested=suggested, continued_run_id=continued_run_id),
+        )
+
+    return _apply
+
+
+@pytest.fixture
+def patch_patched(monkeypatch):
+    """Patch ``workflow.patched`` used inside the BaseWorkflow module."""
+
+    def _apply(value: bool) -> None:
+        monkeypatch.setattr(
+            base_workflow_module.workflow, "patched", lambda _patch_id: value
+        )
+
+    return _apply
+
+
+def test_recycles_when_temporal_suggests(patch_info):
+    patch_info(suggested=True)
+    assert _ConcreteWorkflow().should_continue_as_new() is True
+
+
+def test_no_recycle_when_not_suggested(patch_info):
+    patch_info(suggested=False)
+    assert _ConcreteWorkflow().should_continue_as_new() is False
+
+
+def test_recycling_active_tracks_patched(patch_patched):
+    # Pre-patch in-flight workflow (no marker in history) → inactive.
+    patch_patched(False)
+    assert _ConcreteWorkflow().recycling_active() is False
+    # Patched code → active.
+    patch_patched(True)
+    assert _ConcreteWorkflow().recycling_active() is True
+
+
+def test_is_continued_run_false_on_original_run(patch_info):
+    patch_info(suggested=False, continued_run_id=None)
+    assert _ConcreteWorkflow().is_continued_run() is False
+
+
+def test_is_continued_run_true_after_recycle(patch_info):
+    patch_info(suggested=False, continued_run_id="run-123")
+    assert _ConcreteWorkflow().is_continued_run() is True