Lab 1: The Machine

import asyncio
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from common import TASK_QUEUE, connect, show_history  # noqa: E402
from workflows import AgentRun  # noqa: E402
from temporalio.exceptions import WorkflowAlreadyStartedError


async def main() -> None:
    client = await connect()
    workflow_id = os.environ.get("WORKFLOW_ID", "agent-42")
    try:
        handle = await client.start_workflow(
            AgentRun.run, "summarize the Temporal docs", id=workflow_id, task_queue=TASK_QUEUE
        )
    except WorkflowAlreadyStartedError:
        print(
            f"{workflow_id} is already running. After crashing the Worker, run the Worker again "
            f"to resume this Workflow; do not start {workflow_id} a second time.",
            file=sys.stderr,
        )
        raise SystemExit(2) from None
    print(f"started {handle.id} on {TASK_QUEUE!r} run={handle.result_run_id}; open http://localhost:8233")
    print("result:", await handle.result())
    await show_history(client, handle.id)


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import logging
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from common import TASK_QUEUE, connect  # noqa: E402
from temporalio.worker import Worker  # noqa: E402
from workflows import AgentRun, call_llm  # noqa: E402

# Show the `call_llm key=...` line, so you can count how many times the Activity really ran (1.2).
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")


async def main() -> None:
    client = await connect()
    worker = Worker(
        client,
        task_queue=TASK_QUEUE,
        workflows=[AgentRun],
        activities=[call_llm],
    )
    print(f"worker pid={os.getpid()} polling {TASK_QUEUE!r}; kill -9 {os.getpid()} to crash it", flush=True)
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

"""Lab 1: AgentRun, stage 1 — plan one step, call the model once, sleep, finish.

This is the program the whole course grows. Nothing here is Temporal-magic: an
Activity performs external work, and its outcome becomes part of the Event
History; a Workflow is deterministic code whose Commands, and the results that
came back for them, are recorded, so a fresh Worker can rebuild its state by
replaying them. Temporal never stores `self.step`; it stores enough history to
recompute it.

The Task Queue name is deployment configuration, not Workflow logic, so it lives
in common/ and only worker.py and starter.py read it.
"""
from __future__ import annotations

import asyncio
import hashlib
from dataclasses import dataclass
from datetime import timedelta

from temporalio import activity, workflow


@dataclass
class Step:
    kind: str          # "llm" | "finish"
    prompt: str = ""


@activity.defn
async def call_llm(prompt: str, key: str) -> str:
    """The model call as an Activity. `key` is the idempotency key a retry
    would reuse (module 4 puts a cache behind it). Stage 1 fakes the model."""
    activity.logger.info("call_llm key=%s", key)
    return f"[model answer to: {prompt[:40]}]"


@workflow.defn
class AgentRun:
    def __init__(self) -> None:
        self.step = 0
        self.status = "running"
        self.goal = ""
        self.context_summary = ""

    def plan(self) -> Step:
        """Workflow code: a pure function of state. It never calls a model."""
        if self.step == 0:
            return Step("llm", prompt=f"Make a one-line plan for: {self.goal}")
        return Step("finish")

    @workflow.run
    async def run(self, goal: str, sleep_seconds: int = 30) -> str:
        self.goal = goal
        while True:
            step = self.plan()
            if step.kind == "finish":
                self.status = "done"
                return self.context_summary
            key = f"{workflow.info().workflow_id}:{self.step}:{hashlib.sha256(step.prompt.encode()).hexdigest()[:12]}"
            answer = await workflow.execute_activity(
                call_llm, args=[step.prompt, key], start_to_close_timeout=timedelta(seconds=10)
            )
            self.context_summary = answer
            self.step += 1
            # A durable timer: recorded as TimerStarted; fires as TimerFired; no process waits.
            await asyncio.sleep(sleep_seconds)