Durable Research Workflows with Temporal and Live Search

The rule that shapes everything
Activities
Retry policy where it belongs
The workflow
Why replay actually saves money here
The worker
Where this is and isn’t worth it

Long research jobs fail in boring ways. A deploy rolls the pod, a scrape hangs past its timeout, the process gets OOM-killed on a large page. With a plain script, twenty minutes of completed searches and reads evaporate and you start over — paying for every call twice.

Durable execution fixes this by recording each completed step. On restart, the workflow replays from history: finished activities return their recorded results instantly, and execution resumes at the exact point it stopped.

The rule that shapes everything

Workflow code must be deterministic. It re-executes on every replay, so anything non-deterministic — network calls, clocks, randomness, file I/O — goes in an activity, not the workflow.

For a research pipeline that means every Serply call is an activity, and the workflow is just the plan.

Activities

import os
import requests
from urllib.parse import quote_plus
from dataclasses import dataclass
from temporalio import activity

API_KEY = os.environ["SERPLY_API_KEY"]
BASE = "https://api.serply.io/v1"


@dataclass
class SearchInput:
    query: str
    num: int = 10
    location: str = "US"


@dataclass
class SearchResult:
    query: str
    items: list[dict]


@activity.defn
async def search(inp: SearchInput) -> SearchResult:
    resp = requests.get(
        f"{BASE}/search/q={quote_plus(inp.query)}&num={inp.num}",
        headers={"X-Api-Key": API_KEY, "X-Proxy-Location": inp.location},
        timeout=30,
    )
    if resp.status_code == 429:
        raise RuntimeError("rate_limited")     # retryable
    resp.raise_for_status()

    return SearchResult(
        query=inp.query,
        items=[
            {"title": r.get("title", ""), "url": r.get("link", ""),
             "snippet": r.get("description", ""), "position": r.get("position")}
            for r in resp.json().get("results", [])
        ],
    )


@activity.defn
async def read_page(url: str) -> dict:
    try:
        resp = requests.post(
            f"{BASE}/request",
            headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
            json={"url": url, "response_type": "markdown"},
            timeout=90,
        )
        resp.raise_for_status()
    except requests.HTTPError as e:
        status = e.response.status_code if e.response else None
        if status in (404, 403, 410):
            # Permanent — don't burn retries on it
            raise activity.ApplicationError(
                f"unfetchable: {status}", non_retryable=True
            )
        raise

    return {"url": url, "text": resp.text[:40_000], "chars": len(resp.text)}

Two things worth flagging.

Markdown mode returns text as the response body, so .text rather than .json().

non_retryable=True on 404/403/410 matters more than it looks. Temporal’s default is to retry until the activity’s timeout, and a permanently-gone page will consume the full retry schedule producing the same failure. Classifying it as terminal frees the workflow to move on in seconds instead of minutes.

Retry policy where it belongs

from datetime import timedelta
from temporalio.common import RetryPolicy

SEARCH_RETRY = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(seconds=30),
    maximum_attempts=5,
)

SCRAPE_RETRY = RetryPolicy(
    initial_interval=timedelta(seconds=2),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(seconds=60),
    maximum_attempts=3,
    non_retryable_error_types=["unfetchable"],
)

This is the real payoff over hand-rolled retry code. Exponential backoff, jitter, attempt limits, and terminal-error classification are configuration rather than a utility module you maintain — and they survive a process restart, which your time.sleep() loop does not.

The workflow

import asyncio
from temporalio import workflow


@workflow.defn
class ResearchWorkflow:
    def __init__(self):
        self._status = "starting"
        self._sources: list[str] = []

    @workflow.run
    async def run(self, topic: str, max_sources: int = 8) -> dict:
        self._status = "planning"
        queries = await workflow.execute_activity(
            plan_queries, topic,
            start_to_close_timeout=timedelta(seconds=60),
            retry_policy=SEARCH_RETRY,
        )

        self._status = "searching"
        searches = await asyncio.gather(*[
            workflow.execute_activity(
                search, SearchInput(query=q, num=10),
                start_to_close_timeout=timedelta(seconds=45),
                retry_policy=SEARCH_RETRY,
            )
            for q in queries
        ])

        seen, urls = set(), []
        for s in searches:
            for item in s.items:
                u = item["url"]
                if u and u not in seen:
                    seen.add(u)
                    urls.append(u)
        urls = urls[:max_sources]

        self._status = "reading"
        pages = await asyncio.gather(
            *[
                workflow.execute_activity(
                    read_page, u,
                    start_to_close_timeout=timedelta(seconds=120),
                    retry_policy=SCRAPE_RETRY,
                )
                for u in urls
            ],
            return_exceptions=True,
        )
        good = [p for p in pages if not isinstance(p, BaseException)]
        self._sources = [p["url"] for p in good]

        self._status = "synthesizing"
        report = await workflow.execute_activity(
            synthesize, {"topic": topic, "pages": good},
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=2),
        )

        self._status = "done"
        return {"topic": topic, "report": report, "sources": self._sources,
                "attempted": len(urls), "succeeded": len(good)}

    @workflow.query
    def status(self) -> dict:
        return {"phase": self._status, "sources": len(self._sources)}

asyncio.gather inside a workflow is fine and encouraged — Temporal’s deterministic event loop schedules the activities concurrently and records their completions in a stable order.

return_exceptions=True on the scrape fan-out is the difference between a partial result and no result. One dead URL out of eight should cost you one source, not the whole job.

The @workflow.query handler gives you live progress on a running workflow without any extra infrastructure — you can poll it from a status endpoint while the job is mid-flight.

Why replay actually saves money here

Say the worker dies during the reading phase, after five of eight pages. On restart, Temporal replays the workflow: plan_queries returns its recorded output, all the searches return theirs, and the five completed reads return theirs — none of these re-hit the API. Execution resumes at the three unfinished reads.

That property is the whole reason to reach for this. On a pipeline making dozens of paid API calls, restart-from-scratch is a real cost line, and it’s the one that shows up on a bad deploy day.

The worker

from temporalio.client import Client
from temporalio.worker import Worker
from concurrent.futures import ThreadPoolExecutor


async def main():
    client = await Client.connect("localhost:7233")
    with ThreadPoolExecutor(max_workers=20) as pool:
        worker = Worker(
            client,
            task_queue="research",
            workflows=[ResearchWorkflow],
            activities=[plan_queries, search, read_page, synthesize],
            activity_executor=pool,
        )
        await worker.run()

The requests calls above are synchronous, so they need a thread pool executor rather than blocking the async worker loop. If you’d rather stay fully async, swap requests for httpx.AsyncClient and drop the executor — either works, but mixing them silently starves the worker.

Where this is and isn’t worth it

Not for a chat agent answering in five seconds. The operational cost of running a Temporal cluster dwarfs the benefit when a failure just means the user retries.

It earns its place when a single job runs for minutes, makes dozens of billable calls, and someone is waiting on the output — scheduled competitive reports, bulk enrichment, overnight monitoring sweeps. There, “resume where it stopped” is the difference between a retry that costs nothing and one that costs the whole run again.

Research Workflows That Survive a Deploy