Durable AI Agent dengan Temporal: Tahan Crash dan Penantian

Bayangkan agent purchase request di sebuah ERP. Agent membaca permintaan, mengecek sisa budget cost centre, menulis ringkasan untuk manajer departemen, lalu menunggu keputusan. Manajernya sedang di lokasi supplier sampai Kamis. Selasa malam tim men-deploy rilis baru, pod worker restart, dan await di memori yang menyimpan posisi agent ikut hilang. Tidak ada yang sadar sampai requester bertanya kenapa PO-nya tidak pernah dibuat.

Itulah masalah yang diselesaikan durable AI agent dengan Temporal: posisi agent disimpan oleh Temporal server, bukan di sebuah proses, sehingga crash, restart, atau penantian tiga hari tetap lanjut di langkah yang sama. Artikel ini menjelaskan pembagian workflow dan activity, mengimplementasikan approval pembelian dengan plugin temporalio-openai-agents untuk OpenAI Agents SDK, menunjukkan cara merilis versi baru saat approval masih terbuka, dan membandingkan pekerjaan yang sama di Pydantic AI dan LangGraph. Semua nama API di bawah diambil dari dokumentasi resmi Temporal, Pydantic, OpenAI, dan LangChain yang tercantum di akhir.

Apa yang sebenarnya dijamin durable AI agent di Temporal

Satu run agent adalah loop model call dan tool call, dan setiap langkah bergantung pada hasil langkah sebelumnya. Kalau loop itu hidup di coroutine Python, umurnya sama dengan umur proses. Jika worker crash setelah tool call ketiga, retry yang naif mulai dari awal: memanggil model lagi, mendapat rencana yang sedikit berbeda, membaca ERP lagi, dan bisa menulis dua kali. Jika loop harus menunggu manusia, prosesnya harus tetap hidup selama orang itu butuh waktu, dan tidak ada yang bisa menjanjikan itu.

Jaminan Temporal lebih sempit tetapi lebih berguna daripada sekadar retry. Selama workflow berjalan, server mencatat setiap activity yang dijadwalkan beserta hasilnya di event history workflow. Saat sebuah worker mati, worker lain memutar ulang kode workflow terhadap history itu: model call dan tool call yang sudah selesai mengembalikan hasil yang tercatat tanpa dijalankan lagi, lalu eksekusi berlanjut dari langkah pertama yang belum punya hasil. Penantian berupa timer atau kondisi di server, jadi workflow yang diam tiga hari sama sekali tidak memakai worker.

Pembagian workflow dan activity, langkah demi langkah

Semuanya bertumpu pada satu aturan dari dokumentasi Temporal: kode workflow harus deterministik karena dieksekusi ulang saat replay, sedangkan activity boleh melakukan I/O apa pun tetapi diulang dari awal jika gagal di tengah jalan. Untuk agent, aturan ini memilah setiap baris kode ke salah satu dari dua tempat.

Baris ketiga adalah keputusan desain paling penting di ERP. Model boleh membaca dan meringkas, tetapi langkah yang membuat purchase order dipanggil oleh workflow setelah seseorang menyetujuinya. Dengan begitu aksi yang tidak bisa dibatalkan tetap di luar kendali model, dan ada satu tempat yang jelas untuk idempotency key.

Implementasi agent approval dengan temporalio-openai-agents

Integrasinya sekarang dirilis sebagai package tersendiri. Dokumentasi Temporal menyebut versi 1.0.0 memindahkannya keluar dari Python SDK: extra temporalio[openai-agents] menjadi distribusi temporalio-openai-agents, dan import pindah dari temporalio.contrib.openai_agents ke temporalio.openai_agents. Package ini butuh Python 3.10 ke atas dan Temporal Python SDK 1.33.0 ke atas. Mulai dari activity, yaitu activity Temporal biasa tanpa kode agent di dalamnya.

# uv add temporalio-openai-agents
# (1.0.0 moved the plugin out of temporalio.contrib.openai_agents)

# activities.py: everything that touches the network lives here
from dataclasses import dataclass

from temporalio import activity


@dataclass
class PurchaseRequest:
    pr_number: str
    requester: str
    total_idr: int
    cost_center: str


@activity.defn
async def get_purchase_request(pr_number: str) -> PurchaseRequest:
    """Read one purchase request from the ERP. Read-only."""
    return await erp.fetch_pr(pr_number)


@activity.defn
async def get_budget_remaining(cost_center: str) -> int:
    """Remaining budget for a cost centre this period, in rupiah."""
    return await erp.budget_remaining(cost_center)


@activity.defn
async def notify_approver(pr_number: str, summary: str) -> None:
    await chat.send_to_manager(pr_number, summary)


@activity.defn
async def convert_pr_to_po(pr_number: str, approver: str) -> str:
    # A retried activity runs again from the top, so the ERP call takes
    # the PR number as its idempotency key and returns the existing PO.
    return await erp.create_po_from_pr(
        pr_number, approved_by=approver, idempotency_key=pr_number
    )

Workflow memegang agent. Di dalamnya Anda menulis kode OpenAI Agents SDK biasa, karena plugin mengarahkan Runner.run sehingga setiap model call menjadi activity; tidak ada runner khusus Temporal. Tool yang melakukan I/O dibungkus dengan activity_as_tool, yang mengekspos activity ke model sebagai tool dan menjadwalkan activity itu setiap kali model memanggilnya.

# workflows.py: deterministic orchestration, no I/O
import asyncio
from dataclasses import dataclass
from datetime import timedelta

from temporalio import workflow
from temporalio.openai_agents.workflow import activity_as_tool

with workflow.unsafe.imports_passed_through():
    from agents import Agent, Runner
    from activities import (
        convert_pr_to_po,
        get_budget_remaining,
        get_purchase_request,
        notify_approver,
    )


@dataclass
class Decision:
    approver: str
    approved: bool
    note: str = ""


APPROVAL_WINDOW = timedelta(days=3)


@workflow.defn
class PurchaseApprovalWorkflow:
    def __init__(self) -> None:
        self.decision: Decision | None = None

    @workflow.signal
    def decide(self, decision: Decision) -> None:
        self.decision = decision

    @workflow.query
    def status(self) -> str:
        return "decided" if self.decision else "waiting for approver"

    @workflow.run
    async def run(self, pr_number: str) -> str:
        reviewer = Agent(
            name="pr-reviewer",
            instructions=(
                "Summarise the purchase request for a manager in Bahasa Indonesia: "
                "total, cost centre, and whether budget remains. "
                "Do not recommend approval or rejection."
            ),
            tools=[
                activity_as_tool(get_purchase_request,
                                 start_to_close_timeout=timedelta(seconds=15)),
                activity_as_tool(get_budget_remaining,
                                 start_to_close_timeout=timedelta(seconds=15)),
            ],
        )
        # Ordinary Agents SDK code. The plugin turns every model call into an
        # activity, and activity_as_tool does the same for each tool call, so
        # a replay after a crash reads their results from history.
        review = await Runner.run(reviewer, input=f"Review {pr_number}")

        await workflow.execute_activity(
            notify_approver,
            args=[pr_number, review.final_output],
            start_to_close_timeout=timedelta(seconds=30),
        )

        try:
            # A durable timer on the Temporal server. No worker holds this
            # wait in memory, so it survives restarts and deploys.
            await workflow.wait_condition(
                lambda: self.decision is not None, timeout=APPROVAL_WINDOW
            )
        except asyncio.TimeoutError:
            return f"{pr_number}: no decision in {APPROVAL_WINDOW.days} days, escalated"

        if not self.decision.approved:
            return f"{pr_number}: rejected by {self.decision.approver}"

        po_number = await workflow.execute_activity(
            convert_pr_to_po,
            args=[pr_number, self.decision.approver],
            start_to_close_timeout=timedelta(seconds=60),
        )
        return f"{pr_number}: approved, {po_number} created"

Tiga detail menopang durability-nya. Pembacaan oleh agent lewat activity, jadi crash setelah get_budget_remaining memutar ulang nilai budget yang tercatat tanpa membacanya lagi. Penantiannya adalah wait_condition dengan timeout tiga hari, yang menjadi timer di sisi server. Dan PO dibuat oleh workflow hanya setelah signal membawa keputusan, dengan nomor PR sebagai idempotency key di ERP sehingga activity yang di-retry mengembalikan PO yang sudah dibuat.

Menunggu manajer: signal, query, dan worker

Worker dan setiap client yang memulai atau mengirim signal ke workflow harus memakai OpenAIAgentsPlugin yang sama, karena plugin juga mengatur data converter Pydantic yang men-serialize payload di kedua sisi. ModelActivityParameters mengatur activity model: start_to_close_timeout default-nya 60 detik, dan parameter ini juga menerima retry policy, heartbeat timeout, dan task queue. activity_as_tool hanya menjelaskan cara agent memanggil activity, jadi fungsi activity tetap harus didaftarkan di worker.

# worker.py: the same plugin goes on every Client, worker and caller alike
from datetime import timedelta

from temporalio.client import Client
from temporalio.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin
from temporalio.worker import Worker

PLUGIN = OpenAIAgentsPlugin(
    model_params=ModelActivityParameters(start_to_close_timeout=timedelta(seconds=60))
)

client = await Client.connect("localhost:7233", plugins=[PLUGIN])
worker = Worker(
    client,
    task_queue="erp-approvals",
    workflows=[PurchaseApprovalWorkflow],
    # activity_as_tool does not register activities; list them here too.
    activities=[get_purchase_request, get_budget_remaining,
                notify_approver, convert_pr_to_po],
)
await worker.run()

# api.py: one workflow per PR, and the PR number is the workflow id
await client.start_workflow(
    PurchaseApprovalWorkflow.run,
    "PR-2026-10-0042",
    id="pr-approval-PR-2026-10-0042",
    task_queue="erp-approvals",
)

# Two days later, from the manager's Approve button, in any process:
handle = client.get_workflow_handle("pr-approval-PR-2026-10-0042")
await handle.signal(
    PurchaseApprovalWorkflow.decide,
    Decision(approver="budi.santoso", approved=True),
)

Memakai nomor PR sebagai workflow id memberi approval sebuah business key. Tombol approve hanya butuh id itu untuk menemukan eksekusi yang sedang menunggu, dan query status membuat layar ERP bisa menampilkan waiting for approver tanpa menyentuh state workflow. Jika pemanggil perlu menerima hasil dari panggilan yang sama, Temporal juga punya Update, yang dipakai dokumentasi integrasi untuk giliran chat yang mengembalikan balasan agent; untuk approval yang cukup dikirim lalu ditinggal, signal sudah cukup.

Deploy saat approval masih terbuka

Durability membawa kewajiban. Workflow yang dimulai minggu lalu akan di-replay oleh kode minggu ini, jadi setiap perubahan yang mengubah urutan langkah yang akan diambil eksekusi yang sedang berjalan akan merusak replay-nya. Panduan versioning Temporal menawarkan dua alat: patched(), yang mencatat marker di history sehingga eksekusi lama tetap di jalur lama sementara eksekusi baru mengambil jalur baru, dan Worker Versioning, yang mengunci eksekusi ke versi deployment kode worker yang memulainya.

# The new release adds a vendor blacklist check before the approval wait.

# Wrong: insert the step directly. A workflow that started last week and
# is parked in wait_condition replays a history with no such activity,
# and the worker fails that workflow task with a non-determinism error.
await workflow.execute_activity(
    check_vendor_blacklist, pr_number, start_to_close_timeout=timedelta(seconds=15)
)

# Right: gate the new step behind a patch marker. Old histories take the
# old path; new executions record the marker and take the new one.
if workflow.patched("vendor-blacklist-check"):
    await workflow.execute_activity(
        check_vendor_blacklist, pr_number, start_to_close_timeout=timedelta(seconds=15)
    )

# Once no pre-patch execution is still open, swap patched() for
# workflow.deprecate_patch("vendor-blacklist-check"), then remove it.

Siklus hidup patch punya tiga langkah, dan panduannya tegas soal urutannya: deploy dengan patched(), ganti ke deprecate_patch() setelah tidak ada lagi eksekusi pra-patch yang terbuka, lalu hapus keduanya setelah history tersebut keluar dari masa retention. Untuk workflow approval dengan timeout berhari-hari, itu berarti beberapa kali deploy, bukan beberapa jam. Temporal juga menyarankan replay testing, yaitu menjalankan history yang tercatat terhadap kode baru sebelum dirilis, untuk menemukan perubahan yang butuh patch.

Agent yang sama di Pydantic AI dengan TemporalDurability

Pydantic AI mendukung Temporal secara native dan mengikuti pembagian yang sama: model request, tool call yang mungkin butuh I/O, dan komunikasi dengan MCP server menjadi activity, sedangkan run agent hidup di workflow. API terbarunya berupa capability. Anda memasang TemporalDurability pada Agent biasa, mendaftarkan agent di subclass PydanticAIWorkflow, dan terhubung dengan PydanticAIPlugin. Wrapper TemporalAgent yang lama sudah deprecated dan akan dihapus di v3; dokumentasinya menyebut workflow yang dimulai dengannya tetap di-replay dengan benar setelah beralih, selama nama agent, id toolset, dan key registry model tidak berubah.

# uv add "pydantic-ai[temporal]"
from datetime import timedelta

from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import (
    PydanticAIPlugin,
    PydanticAIWorkflow,
    TemporalDurability,
)
from temporalio import workflow
from temporalio.workflow import ActivityConfig

reviewer = Agent(
    "openai:gpt-5.6-sol",
    # Required under Temporal: activity names derive from the agent name.
    # Renaming it in production breaks every workflow still in flight.
    name="pr-reviewer",
    instructions="Summarise the purchase request for a manager. Do not recommend.",
    tools=[get_purchase_request, get_budget_remaining],  # plain async functions
    capabilities=[
        TemporalDurability(
            activity_config=ActivityConfig(start_to_close_timeout=timedelta(seconds=60))
        )
    ],
)


@workflow.defn
class PurchaseApprovalWorkflow(PydanticAIWorkflow):
    __pydantic_ai_agents__ = [reviewer]  # the plugin registers their activities

    @workflow.run
    async def run(self, pr_number: str) -> str:
        # Must be the async API: run_sync() inside a workflow raises UserError.
        review = await reviewer.run(f"Review {pr_number}")
        ...  # same notify / wait_condition / convert_pr_to_po steps as before


client = await Client.connect("localhost:7233", plugins=[PydanticAIPlugin()])

Dua perbedaan dari plugin OpenAI menentukan bentuk kodenya. Function tool async biasa otomatis dialihkan lewat activity, jadi tidak ada langkah activity_as_tool. Sebagai gantinya, semua yang menyeberang ke activity harus bisa di-serialize oleh Pydantic, termasuk deps, model_settings, dan metadata tool, dan Temporal membatasi setiap payload 2 MB secara default. Timeout activity default-nya 60 detik kecuali Anda mengisi activity_config, dan dokumentasinya menyarankan mematikan retry bawaan client provider, misalnya max_retries diisi 0, agar retry policy Temporal tidak dikalikan dua lapis retry lainnya.

Temporal, checkpoint LangGraph, atau RunState yang disimpan

Temporal bukan satu-satunya cara membuat agent hidup lebih lama dari prosesnya. LangGraph menyimpan state graph per thread lewat checkpointer, yang dokumentasinya sebut untuk human-in-the-loop dan fault tolerance, dan Agents SDK bisa men-serialize run yang sedang pause dengan sendirinya. Perbedaannya ada pada apa yang disimpan, seberapa sering, dan apa yang harus Anda operasikan.

Dalam praktiknya, pilihannya mengikuti bentuk pekerjaan. Jika agent hanyalah satu langkah dalam proses bisnis yang memang butuh timer, retry, dan audit history, Temporal memberi semuanya dari satu sistem. Jika agent itu sendiri adalah prosesnya dan berjalan di graph yang sudah Anda kelola, checkpointer LangGraph dalam mode sync lebih ringan dioperasikan. Snapshot RunState saja cocok untuk satu jeda approval, tetapi dokumentasi OpenAI memperingatkan bahwa snapshot itu tidak diautentikasi: muat hanya dari storage yang Anda percaya, karena isinya membawa argumen tool.

# LangGraph: durability is a per-call setting on a checkpointed graph.
#   "exit"  saves only when the run exits or interrupts: a crash mid-run loses it
#   "async" saves while the next step runs: a small window can be lost
#   "sync"  saves before the next step starts: the one to use for ERP writes
graph.invoke(inputs, {"configurable": {"thread_id": "PR-2026-10-0042"}}, durability="sync")

# OpenAI Agents SDK without Temporal: a paused run is a JSON snapshot you own.
state = result.to_state()
db.save("PR-2026-10-0042", state.to_json())     # pending tool call + arguments
# ...days later, in another process...
state = await RunState.from_json(agent, db.load("PR-2026-10-0042"))
state.approve(state.get_interruptions()[0])
result = await Runner.run(agent, state)

Checklist produksi untuk durable agent

Durable execution menghilangkan satu kelas kegagalan dan membuat beberapa kesalahan lain lebih mudah terjadi. Sebelum agent seperti ini live di ERP:

1. Beri setiap activity tulis sebuah idempotency key dari dokumen bisnisnya, misalnya nomor PR. Activity di-retry dari awal, dan timeout tidak membuktikan percobaan pertama gagal.
2. Atur start_to_close_timeout untuk satu kali round trip, bukan untuk seluruh pekerjaan, dan batasi retry activity model dengan sengaja: Pydantic AI mencatat bahwa maximum attempts default Temporal bernilai 0, artinya eksekusi ulang tanpa batas.
3. Jauhkan dokumen besar dari payload workflow. Kirim nomor PR dan biarkan activity membaca lampirannya, atau simpan sendiri lalu kirim referensinya, agar tetap di bawah default 2 MB.
4. Untuk agent bergaya chat dengan banyak giliran, pakai Continue-as-New sebelum event history terlalu besar, dengan membawa result.to_input_list() dan agent saat ini ke eksekusi baru seperti contoh di dokumentasi Temporal.
5. Jalankan replay test terhadap history yang tercatat di CI, dan lindungi setiap perubahan urutan langkah dengan patched() atau Worker Versioning.

Aturan yang perlu dibawa sederhana: apa pun yang tidak boleh dilupakan agent masuk ke history, dan apa pun yang tidak boleh diulang agent masuk ke activity dengan idempotency key. Dengan model call dan tool di activity, penantian di timer sisi server, dan langkah baru di balik patch marker, approval pembelian bisa menunggu tiga hari melewati dua kali deploy dan tetap lanjut tepat di langkah tempat ia berhenti.