Loops & advanced workflows Preview

Preview

The worker repository referenced below is still hardening before its OSS release. The patterns and code are accurate to what the team uses internally today.

This guide builds on Swap in a real agent. You should have the full stack running with the real Strands Base Agent on localhost:8081 before continuing.

Prerequisite check — all four services healthy:

curl -s -o /dev/null -w "HTTP %{http_code}\n" http://localhost:8080        # Temporal UI
curl -s -o /dev/null -w "HTTP %{http_code}\n" http://localhost:8000/health/ # Workflow API
curl -s -o /dev/null -w "HTTP %{http_code}\n" http://localhost:8081/api/v1/health  # Strands Base Agent

By the end you'll have replaced the simple 3-step MyWorkflow with a multi-phase, loop-based workflow that demonstrates:

• Multi-phase orchestration — sequential workflow phases with distinct loop logic
• Human-in-the-loop (HIL) feedback loops — while loops that pause for signals and iterate on revisions
• Iteration tracking — versioned document keys and counters
• Multiple signal & query handlers — per-phase approve/feedback signals and real-time status queries
• Timeout handling — configurable deadlines inside loops
• Richer data models — Pydantic models with validators

---

Architecture: simple vs. advanced

Before — linear 3-step workflow

Start ──▶ my_activity ──▶ call_strands_agent ──▶ wait for approval ──▶ End
                          (Strands Base Agent
                           via host.docker.internal:8081)

Three steps in sequence with a single approval gate at the end.

After — multi-phase loop workflow

Start ──▶ ┌─────────────────────────────────┐
          │  Phase 1: Generate Plan          │
          │  ┌───────────────────────────┐   │
          │  │ call_strands_agent        │   │
          │  └───────────┬───────────────┘   │
          │              ▼                   │
          │  ┌───────────────────────────┐   │
          │  │ Wait for signal           │◀─┐│
          │  │ (approve or feedback)     │  ││
          │  └───────┬──────────┬────────┘  ││
          │  approved│          │feedback   ││
          │          │          ▼           ││
          │          │  ┌───────────────┐  ││
          │          │  │ Revise plan   │──┘│
          │          │  │ (agent call)  │   │
          │          │  └───────────────┘   │
          └──────────┼─────────────────────┘
                     ▼
          ┌─────────────────────────────────┐
          │  Phase 2: Generate Research      │
          │  (same loop pattern)             │
          └──────────┬──────────────────────┘
                     ▼
                    End

Each phase runs an initial agent call, then enters a while loop that waits for a human signal — either approval (exits the loop) or feedback (triggers a revision and loops again).

---

What you'll build

Component	File	Description
Workflow	src/workflows.py	ResearchWorkflow with generate_plan_loop and generate_research_loop
Activities	src/activities.py	call_strands_agent with task-based routing
Models	src/models.py	ResearchRequest, ResearchResult, ModelValidation
Config	config/workflows.yaml	Workflow registration
Config	config/activities.yaml	Activity registration with retry policy

---

Step 1 — The loop pattern

The core pattern is a signal-driven while loop inside a workflow method:

async def generate_plan_loop(self, request):
    """Generate plan, then loop until human approves."""
    self.plan_iterations = 1

    # 1. Initial generation (activity call)
    await execute_activity("call_strands_agent", args=[...])

    # 2. HIL approval loop
    while not self.plan_approved:
        # Block until a signal arrives
        await workflow.wait_condition(
            lambda: self.is_plan_decision_received(),
            timeout=timedelta(hours=24),
        )

        if self.plan_feedback_received and not self.plan_approved:
            # Feedback received — revise and loop again
            self.plan_iterations += 1
            await execute_activity("call_strands_agent", args=[...])
            self.reset_plan_feedback()

    # Loop exits when plan_approved == True

Key mechanics:

1. workflow.wait_condition blocks execution until the lambda returns True — a signal handler sets the state that flips it.
2. HILSignalsMixin (from foundry_temporal.signals) provides the state and helper methods: plan_approved, plan_feedback_received, plan_feedback_comments, document_approved, etc.
3. reset_plan_feedback() clears feedback state so the loop can wait for the next decision.
4. Iteration counters drive versioned document keys like research-abc123-plan-v2.

Temporal guarantee

Even if the worker crashes mid-loop, Temporal replays the workflow from its event history. The while loop, counters, and signal state are all reconstructed deterministically.

---

Step 2 — Data models

Replace src/models.py:

"""
Advanced Workflow Models — Multi-phase research workflow with loop support.
"""

from typing import Any

from pydantic import BaseModel, ConfigDict, Field, field_validator


class ResearchRequest(BaseModel):
    """Input model for the research workflow."""

    workflow_id: str = Field(..., description="Unique workflow identifier")
    topic: str = Field(
        ..., min_length=10, max_length=500, description="Research topic or question"
    )
    criteria: dict[str, Any] | str = Field(
        ...,
        description="Research criteria — structured dict or natural language",
    )
    agent_url: str = Field(
        default="http://host.docker.internal:8081",
        description="Strands agent endpoint URL",
    )

    @field_validator("topic")
    @classmethod
    def validate_topic(cls, v: str) -> str:
        if len(v.strip()) < 10:
            raise ValueError("Topic must be at least 10 characters long")
        return v.strip()

    @field_validator("workflow_id")
    @classmethod
    def validate_workflow_id(cls, v: str) -> str:
        if not v.startswith("research-"):
            raise ValueError('Workflow ID must start with "research-"')
        return v

    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "workflow_id": "research-abc123",
                    "topic": "Workflow orchestration platforms comparison",
                    "criteria": {
                        "must_have": ["Durable execution", "Observability"],
                        "should_have": ["Kubernetes-native deployment"],
                    },
                    "agent_url": "http://host.docker.internal:8081",
                }
            ]
        }
    )


class ResearchResult(BaseModel):
    """Output model from completed research workflow."""

    workflow_id: str = Field(..., description="Unique workflow identifier")
    final_document_key: str = Field(..., description="Key for the final approved document")
    iterations: int = Field(..., description="Total revision iterations")
    status: str = Field(..., description="Final status: completed, failed, timeout")
    metadata: dict[str, Any] = Field(..., description="Execution metadata")

    @field_validator("status")
    @classmethod
    def validate_status(cls, v: str) -> str:
        allowed = ["completed", "failed", "timeout"]
        if v not in allowed:
            raise ValueError(f"Status must be one of: {allowed}")
        return v


class ModelValidation:
    """Utility class for document key management across loop iterations."""

    @staticmethod
    def create_document_key(workflow_id: str, doc_type: str, version: int) -> str:
        """Create a versioned document key: {workflow_id}-{type}-v{version}"""
        if doc_type not in ["plan", "results"]:
            raise ValueError(f"Invalid document type: {doc_type}")
        if version < 1:
            raise ValueError(f"Version must be positive: {version}")
        return f"{workflow_id}-{doc_type}-v{version}"

    @staticmethod
    def is_valid_document_key(document_key: str) -> bool:
        parts = document_key.split("-")
        if len(parts) < 3:
            return False
        if not parts[-1].startswith("v") or not parts[-1][1:].isdigit():
            return False
        if parts[-2] not in ["plan", "results"]:
            return False
        return int(parts[-1][1:]) > 0

What changed from the simple setup

• Validated workflow_id, topic with length constraints (was: plain strings).
• field_validator for format enforcement (was: none).
• ResearchResult with status validation and rich metadata (was: flat dict).
• ModelValidation utility for iteration-versioned document keys (new concept).
• Agent URL passed per-request, defaulting to host.docker.internal:8081 (was: looked up from services.yaml).

---

Step 3 — Activities

Replace src/activities.py. The advanced pattern routes different task types through a single call_strands_agent activity:

"""
Advanced Activities — Task-routed agent calls with validation.
"""

from datetime import timedelta
from typing import Any

from temporalio import activity
from temporalio.common import RetryPolicy
from temporalio.exceptions import ApplicationError

# More attempts, longer intervals for agent operations
AGENT_RETRY_POLICY = RetryPolicy(
    initial_interval=timedelta(seconds=2),
    maximum_interval=timedelta(seconds=30),
    maximum_attempts=5,
    backoff_coefficient=2.0,
)


@activity.defn
async def call_strands_agent(task_request: dict[str, Any]) -> None:
    """
    Call Strands agent via A2A protocol with task-based routing.

    The workflow passes a task_request dict whose "task" field selects the
    operation: analyze_and_plan, revise_plan, conduct_research, or
    incorporate_feedback. One activity, four operations.
    """
    import aiohttp

    task = task_request["task"]
    workflow_id = task_request["workflow_id"]
    agent_url = task_request["agent_url"]
    document_key = task_request["document_key"]

    activity.logger.info(f"Calling agent for task: {task}, workflow: {workflow_id}")
    activity.heartbeat(f"Starting agent call: {task}")

    if task == "analyze_and_plan":
        topic = task_request["topic"]
        criteria = task_request["criteria"]
        task_message = (
            f"Analyze and plan research for topic: {topic}. "
            f"Criteria: {criteria}. "
            f"Store plan using document key: {document_key}."
        )
    elif task == "revise_plan":
        feedback = task_request["feedback"]
        existing_key = task_request["existing_document_key"]
        task_message = (
            f"Read existing plan from key: {existing_key}. "
            f"Revise based on feedback: {feedback}. "
            f"Store revised plan using key: {document_key}."
        )
    elif task == "conduct_research":
        topic = task_request["topic"]
        criteria = task_request["criteria"]
        task_message = (
            f"Conduct research on topic: {topic}. "
            f"Criteria: {criteria}. "
            f"Store results using key: {document_key}."
        )
    elif task == "incorporate_feedback":
        feedback = task_request["feedback"]
        existing_key = task_request["existing_document_key"]
        task_message = (
            f"Read existing document from key: {existing_key}. "
            f"Incorporate feedback: {feedback}. "
            f"Store updated document using key: {document_key}."
        )
    else:
        raise ApplicationError(f"Unknown task type: '{task}'", non_retryable=True)

    message_id = f"msg-{workflow_id}-{task}-{activity.info().activity_id}"
    payload = {
        "jsonrpc": "2.0",
        "id": "1",
        "method": "message/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": task_message}],
                "messageId": message_id,
            }
        },
    }

    timeout = aiohttp.ClientTimeout(total=900)
    async with (
        aiohttp.ClientSession(timeout=timeout) as session,
        session.post(agent_url, json=payload) as response,
    ):
        if response.status != 200:
            response_text = await response.text()
            raise ApplicationError(
                f"Agent HTTP error {response.status}: {response_text[:500]}",
                non_retryable=True,
            )
        response_data = await response.json()

    if "error" in response_data:
        raise ApplicationError(
            f"Agent JSON-RPC error: {response_data['error']}",
            non_retryable=True,
        )

    activity.logger.info(
        f"Agent completed {task} successfully, document key: {document_key}"
    )

What changed

• Single task-routed activity, handling 4 operations via the task field (was: two separate activities).
• Custom AGENT_RETRY_POLICY — 5 attempts, exponential backoff (was: default).
• Heartbeats before long-running calls (was: none).
• Returns None — the agent persists results externally via the document key (was: returns result data).
• Agent URL passed in task_request (was: looked up from services.yaml).

---

Step 4 — Workflow

Replace src/workflows.py. This is the core of the advanced pattern:

"""
Advanced Workflow — Multi-phase research with HIL feedback loops.
"""

from datetime import timedelta
from typing import Any

from foundry_temporal.signals import HILSignalsMixin
from temporalio import workflow
from temporalio.common import RetryPolicy
from temporalio.workflow import execute_activity

from .activities import AGENT_RETRY_POLICY
from .models import ModelValidation, ResearchRequest, ResearchResult


@workflow.defn
class ResearchWorkflow(HILSignalsMixin):  # type: ignore[misc]
    """
    Multi-phase HIL Research Workflow with feedback loops.

    Phases:
    1. Plan Generation        — agent creates initial research plan
    2. Plan Review Loop       — human approves or sends feedback (loops until approved)
    3. Research Execution     — agent conducts research based on approved plan
    4. Document Review Loop   — human approves or sends feedback (loops until approved)
    5. Finalization           — return results with iteration metadata
    """

    def __init__(self) -> None:
        super().__init__()
        self.plan_iterations = 0
        self.document_iterations = 0
        self.current_phase = "planning"
        self.processing_status = "idle"

    @workflow.run
    async def run(self, request: ResearchRequest) -> ResearchResult:
        workflow_id = workflow.info().workflow_id
        workflow.logger.info(
            "Starting ResearchWorkflow",
            extra={"workflow_id": workflow_id, "topic": request.topic},
        )

        await self.generate_plan_loop(request)
        await self.generate_research_loop(request)

        final_plan_key = ModelValidation.create_document_key(
            request.workflow_id, "plan", self.plan_iterations
        )
        final_document_key = ModelValidation.create_document_key(
            request.workflow_id, "results", self.document_iterations
        )

        return ResearchResult(
            workflow_id=request.workflow_id,
            final_document_key=final_document_key,
            iterations=self.plan_iterations + self.document_iterations,
            status="completed",
            metadata={
                "plan_iterations": self.plan_iterations,
                "document_iterations": self.document_iterations,
                "approved_plan_key": final_plan_key,
                "final_document_key": final_document_key,
                "completed_at": workflow.now().isoformat(),
            },
        )

    # ── Phase 1+2: Plan generation loop ─────────────────────────────

    async def generate_plan_loop(self, request: ResearchRequest) -> None:
        workflow.logger.info("Starting plan generation loop")
        self.plan_iterations = 1

        plan_document_key = ModelValidation.create_document_key(
            request.workflow_id, "plan", self.plan_iterations
        )

        plan_request = {
            "task": "analyze_and_plan",
            "workflow_id": request.workflow_id,
            "agent_url": request.agent_url,
            "topic": request.topic,
            "criteria": request.criteria,
            "document_key": plan_document_key,
        }

        self.processing_status = "generating"
        await execute_activity(
            "call_strands_agent",
            args=[plan_request],
            start_to_close_timeout=timedelta(minutes=10),
            retry_policy=self._get_agent_retry_policy(),
        )
        self.processing_status = "idle"
        workflow.logger.info(f"Initial plan generated: {plan_document_key}")

        # ── HIL approval loop ──
        while not self.plan_approved:
            workflow.logger.info(
                f"Awaiting plan approval (iteration {self.plan_iterations})"
            )

            try:
                await workflow.wait_condition(
                    lambda: self.is_plan_decision_received(),
                    timeout=timedelta(hours=24),
                )
            except TimeoutError:
                raise RuntimeError(
                    f"Plan approval timeout after 24 hours for workflow "
                    f"{request.workflow_id}"
                ) from None

            if self.plan_feedback_received and not self.plan_approved:
                workflow.logger.info(
                    f"Processing plan feedback: {self.plan_feedback_comments}"
                )

                current_plan_key = ModelValidation.create_document_key(
                    request.workflow_id, "plan", self.plan_iterations
                )
                self.plan_iterations += 1
                revised_plan_key = ModelValidation.create_document_key(
                    request.workflow_id, "plan", self.plan_iterations
                )

                feedback_request = {
                    "task": "revise_plan",
                    "workflow_id": request.workflow_id,
                    "agent_url": request.agent_url,
                    "feedback": self.plan_feedback_comments,
                    "criteria": request.criteria,
                    "existing_document_key": current_plan_key,
                    "document_key": revised_plan_key,
                }

                self.processing_status = "revising"
                await execute_activity(
                    "call_strands_agent",
                    args=[feedback_request],
                    start_to_close_timeout=timedelta(minutes=10),
                    retry_policy=self._get_agent_retry_policy(),
                )
                self.processing_status = "idle"
                self.reset_plan_feedback()
                workflow.logger.info(f"Plan revised: {revised_plan_key}")

        workflow.logger.info(
            f"Plan approved after {self.plan_iterations} iterations"
        )

    # ── Phase 3+4: Research generation loop ──────────────────────────

    async def generate_research_loop(self, request: ResearchRequest) -> None:
        workflow.logger.info("Starting research generation loop")
        self.document_iterations = 1
        self.current_phase = "research"

        results_document_key = ModelValidation.create_document_key(
            request.workflow_id, "results", self.document_iterations
        )

        research_request = {
            "task": "conduct_research",
            "workflow_id": request.workflow_id,
            "agent_url": request.agent_url,
            "topic": request.topic,
            "criteria": request.criteria,
            "document_key": results_document_key,
        }

        self.processing_status = "generating"
        await execute_activity(
            "call_strands_agent",
            args=[research_request],
            start_to_close_timeout=timedelta(minutes=30),
            retry_policy=self._get_agent_retry_policy(),
        )
        self.processing_status = "idle"
        workflow.logger.info(f"Initial research generated: {results_document_key}")

        while not self.document_approved:
            workflow.logger.info(
                f"Awaiting research approval (iteration {self.document_iterations})"
            )

            try:
                await workflow.wait_condition(
                    lambda: self.is_document_decision_received(),
                    timeout=timedelta(hours=24),
                )
            except TimeoutError:
                raise RuntimeError(
                    f"Document approval timeout after 24 hours for workflow "
                    f"{request.workflow_id}"
                ) from None

            if self.document_feedback_received and not self.document_approved:
                current_document_key = ModelValidation.create_document_key(
                    request.workflow_id, "results", self.document_iterations
                )
                self.document_iterations += 1
                revised_results_key = ModelValidation.create_document_key(
                    request.workflow_id, "results", self.document_iterations
                )

                feedback_request = {
                    "task": "incorporate_feedback",
                    "workflow_id": request.workflow_id,
                    "agent_url": request.agent_url,
                    "feedback": self.document_feedback_comments,
                    "criteria": request.criteria,
                    "existing_document_key": current_document_key,
                    "document_key": revised_results_key,
                }

                self.processing_status = "revising"
                await execute_activity(
                    "call_strands_agent",
                    args=[feedback_request],
                    start_to_close_timeout=timedelta(minutes=15),
                    retry_policy=self._get_agent_retry_policy(),
                )
                self.processing_status = "idle"
                self.reset_document_feedback()
                workflow.logger.info(f"Research revised: {revised_results_key}")

        workflow.logger.info(
            f"Research approved after {self.document_iterations} iterations"
        )

    # ── Query handlers ───────────────────────────────────────────────

    @workflow.query
    def get_status(self) -> dict[str, Any]:
        """Real-time workflow status. Doesn't affect execution."""
        workflow_id = workflow.info().workflow_id
        current_plan_key = ""
        current_document_key = ""

        if self.plan_iterations > 0:
            current_plan_key = ModelValidation.create_document_key(
                workflow_id, "plan", self.plan_iterations
            )
        if self.document_iterations > 0:
            current_document_key = ModelValidation.create_document_key(
                workflow_id, "results", self.document_iterations
            )

        return {
            "workflow_id": workflow_id,
            "current_phase": self.current_phase,
            "plan_approved": self.plan_approved,
            "document_approved": self.document_approved,
            "plan_iteration": self.plan_iterations,
            "document_iteration": self.document_iterations,
            "current_plan_key": current_plan_key,
            "current_document_key": current_document_key,
            "processing_status": self.processing_status,
        }

    # ── Signal handlers ──────────────────────────────────────────────

    @workflow.signal
    def approve_plan(self) -> None:
        super().approve_plan()

    @workflow.signal
    def provide_plan_feedback(self, comments: str) -> None:
        super().provide_plan_feedback(comments)

    @workflow.signal
    def approve_research(self) -> None:
        super().approve_research()

    @workflow.signal
    def provide_feedback(self, comments: str) -> None:
        super().provide_feedback(comments)

    # ── Utility ──────────────────────────────────────────────────────

    def _get_agent_retry_policy(self) -> RetryPolicy:
        return AGENT_RETRY_POLICY

Anatomy of the loop

┌─ plan_iterations = 1
│
├─ execute_activity("call_strands_agent", task="analyze_and_plan")
│   └─ Creates document key: research-abc123-plan-v1
│
├─ while not self.plan_approved:
│   │
│   ├─ workflow.wait_condition(is_plan_decision_received, timeout=24h)
│   │   └─ BLOCKS here until a signal arrives
│   │
│   ├─ if feedback received (not approved):
│   │   ├─ plan_iterations += 1  →  now v2
│   │   ├─ execute_activity("call_strands_agent", task="revise_plan")
│   │   │   └─ Reads v1, writes v2
│   │   └─ reset_plan_feedback()  →  clears state for next loop
│   │
│   └─ if approved:
│       └─ while condition is False → loop exits
│
└─ return (plan approved after N iterations)

Signal flow:

• API sends approve_plan → self.plan_approved = True → wait_condition unblocks → loop exits.
• API sends provide_plan_feedback("add more detail") → self.plan_feedback_received = True, self.plan_feedback_comments = "add more detail" → wait_condition unblocks → revision branch executes → reset_plan_feedback() → loop continues.

---

Step 5 — YAML configuration

config/workflows.yaml:

workflows:
  - module: "src.workflows"
    classes: ["ResearchWorkflow"]
    description: "Multi-phase HIL research workflow with plan and document review loops"

config/activities.yaml:

activities:
  - module: "src.activities"
    functions:
      - name: "call_strands_agent"
        description: "Call Strands agent via A2A protocol with task routing"
        retry_policy: "agent"
        timeout_seconds: 900

The agent URL is now passed per-request from the workflow rather than looked up from services.yaml.

---

Step 6 — Rebuild and test

6.1 Rebuild the worker

cd ~/agent-foundry-workspace/temporal-workflow-worker-starter
docker compose build worker
docker compose up -d worker
docker compose logs -f worker

Look for:

• Loaded workflow: ResearchWorkflow from src.workflows
• Loaded activity: call_strands_agent from src.activities
• Temporal worker started successfully

6.2 Start a research workflow

curl -X POST http://localhost:8000/workflows/start \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_type": "ResearchWorkflow",
    "workflow_id": "research-test-001",
    "args": [{
      "workflow_id": "research-test-001",
      "topic": "Compare workflow orchestration platforms for microservices",
      "criteria": {
        "must_have": ["Durable execution", "Observability"],
        "should_have": ["Kubernetes-native", "Multi-language SDKs"]
      },
      "agent_url": "http://host.docker.internal:8081"
    }]
  }'

6.3 Query workflow status

curl -X POST http://localhost:8000/workflows/query \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_id": "research-test-001",
    "query_type": "get_status"
  }'

Expected (waiting for plan approval):

{
  "workflow_id": "research-test-001",
  "current_phase": "planning",
  "plan_approved": false,
  "document_approved": false,
  "plan_iteration": 1,
  "processing_status": "idle"
}

6.4 Send feedback (triggers revision loop)

curl -X POST http://localhost:8000/workflows/signal \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_id": "research-test-001",
    "signal_name": "provide_plan_feedback",
    "signal_data": ["Please add implementation timeline and cost estimates"]
  }'

Query again — iteration count increments:

{ "plan_iteration": 2, "processing_status": "idle" }

6.5 Approve the plan

curl -X POST http://localhost:8000/workflows/signal \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_id": "research-test-001",
    "signal_name": "approve_plan"
  }'

The workflow moves to Phase 3 — research generation.

6.6 Approve the research document

curl -X POST http://localhost:8000/workflows/signal \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_id": "research-test-001",
    "signal_name": "approve_research"
  }'

The workflow completes and returns a ResearchResult.

6.7 Inspect

1. Temporal UI → default namespace → research-test-001. The event history shows:

   • Multiple ActivityTaskScheduled / ActivityTaskCompleted (one per agent call).
   • WorkflowSignalReceived events for each feedback/approval signal.
   • The workflow completing after both loops exit.

2. Strands Base Agent logs — one A2A JSON-RPC request per loop iteration:

cd ~/agent-foundry-workspace/strands-base-agent
docker compose logs -f

---

Key concepts summary

1. Signal-driven loops

while not self.plan_approved:
    await workflow.wait_condition(lambda: self.is_plan_decision_received())
    if self.plan_feedback_received and not self.plan_approved:
        # ... revise and loop
        self.reset_plan_feedback()

The while loop is the core pattern. wait_condition blocks until a signal changes state. The reset call clears state for the next iteration.

2. Versioned document keys

self.plan_iterations += 1
key = ModelValidation.create_document_key(workflow_id, "plan", self.plan_iterations)
# → "research-abc123-plan-v2"

Every iteration increments a counter and generates a new versioned key — full audit trail of every revision.

3. Multiple signal handlers per phase

Signal	Phase	Effect
approve_plan()	Plan	Sets plan_approved = True, exits plan loop
provide_plan_feedback(comments)	Plan	Sets feedback state, triggers plan revision
approve_research()	Research	Sets document_approved = True, exits research loop
provide_feedback(comments)	Research	Sets feedback state, triggers document revision

4. Queries for observability

get_status returns real-time state without affecting execution — useful for dashboards or status endpoints.

5. Timeout handling

await workflow.wait_condition(
    lambda: self.is_plan_decision_received(),
    timeout=timedelta(hours=24),
)

No signal in 24 hours → TimeoutError → caught and converted to a RuntimeError that fails the workflow cleanly.

6. Determinism

Everything inside @workflow.defn must be deterministic:

• No datetime.now() — use workflow.now().
• No direct HTTP calls — delegate to activities via execute_activity.
• No random values — use workflow.random().
• Side effects happen only inside @activity.defn functions.

The while loops, counters, and signal state are replayed deterministically from Temporal's event history if the worker restarts.

---

What's next

• Customize prompts, tools, and capabilities in the agent's .env — see the Strands Base Agent guides.
• For production deployment: worker identity, multi-instance scaling, slot suppliers, graceful shutdown, health checks, and metrics. The companion worker repo ships an HA tuning guide — it'll land here once the OSS release does.
• Temporal documentation — official concepts (signals, queries, activities, workers).