Human-in-the-loop patterns for critical decisions

# temporal-client >= 1.4.0, requires: pip install temporalio structlog
import asyncio
import os
import structlog
from datetime import timedelta
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio.common import RetryPolicy
from temporalio.exceptions import ActivityError, ApplicationError

log = structlog.get_logger()

SIGNAL_HUMAN_DECISION = "human_decision"

@activity.defn
async def run_ai_classifier(document_id: str) -> dict:
    # Replace with real model call; timeout enforced by Temporal schedule_to_close
    import httpx
    api_key = os.environ["OPENAI_API_KEY"]
    async with httpx.AsyncClient(timeout=15.0) as client:
        resp = await client.post(
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "model": "gpt-4o-mini",
                "messages": [{"role": "user", "content": f"Classify document {document_id}: APPROVE or REJECT. Respond with JSON: {{\"decision\": ..., \"confidence\": ...}}"}],
                "response_format": {"type": "json_object"},
            },
        )
        resp.raise_for_status()
        result = resp.json()["choices"][0]["message"]["content"]
        import json
        parsed = json.loads(result)
        log.info("ai_classifier_result", document_id=document_id, decision=parsed["decision"], confidence=parsed["confidence"])
        return parsed

@activity.defn
async def enqueue_human_review(document_id: str, ai_result: dict, workflow_id: str) -> None:
    # Write to a review queue table; reviewer UI reads from here
    import httpx
    webhook = os.environ["REVIEW_QUEUE_WEBHOOK"]
    payload = {
        "workflow_id": workflow_id,
        "document_id": document_id,
        "ai_decision": ai_result["decision"],
        "confidence": ai_result["confidence"],
        "sla_seconds": 3600,
    }
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.post(webhook, json=payload)
        resp.raise_for_status()
    log.info("review_task_enqueued", document_id=document_id, workflow_id=workflow_id)

@activity.defn
async def record_final_decision(document_id: str, decision: str, source: str, reviewer_id: str | None) -> None:
    log.info("final_decision_recorded", document_id=document_id, decision=decision, source=source, reviewer_id=reviewer_id)
    # Persist to audit table here

@workflow.defn
class DocumentApprovalWorkflow:
    def __init__(self):
        self._human_decision: dict | None = None

    @workflow.signal(name=SIGNAL_HUMAN_DECISION)
    async def receive_human_decision(self, payload: dict) -> None:
        log.info("human_signal_received", payload=payload)
        self._human_decision = payload

    @workflow.run
    async def run(self, document_id: str) -> dict:
        retry = RetryPolicy(maximum_attempts=3, initial_interval=timedelta(seconds=2), backoff_coefficient=2.0)

        try:
            ai_result = await workflow.execute_activity(
                run_ai_classifier,
                document_id,
                schedule_to_close_timeout=timedelta(seconds=30),
                retry_policy=retry,
            )
        except ActivityError as e:
            log.error("ai_classifier_failed", document_id=document_id, error=str(e))
            # Graceful degradation: force human review on AI failure
            ai_result = {"decision": "UNKNOWN", "confidence": 0.0}

        CONFIDENCE_THRESHOLD = 0.85
        if ai_result["confidence"] >= CONFIDENCE_THRESHOLD and ai_result["decision"] != "UNKNOWN":
            await workflow.execute_activity(
                record_final_decision,
                document_id, ai_result["decision"], "auto", None,
                schedule_to_close_timeout=timedelta(seconds=10),
            )
            return {"decision": ai_result["decision"], "source": "auto"}

        # Below threshold: pause and wait for human
        await workflow.execute_activity(
            enqueue_human_review,
            document_id, ai_result, workflow.info().workflow_id,
            schedule_to_close_timeout=timedelta(seconds=10),
        )

        # Wait up to 1 hour for human signal; fall back to DEFER on timeout
        try:
            await workflow.wait_condition(lambda: self._human_decision is not None, timeout=timedelta(hours=1))
            decision = self._human_decision["decision"]
            reviewer_id = self._human_decision.get("reviewer_id")
        except asyncio.TimeoutError:
            log.warning("human_review_sla_breached", document_id=document_id)
            decision, reviewer_id = "DEFER", None

        await workflow.execute_activity(
            record_final_decision,
            document_id, decision, "human", reviewer_id,
            schedule_to_close_timeout=timedelta(seconds=10),
        )
        return {"decision": decision, "source": "human", "reviewer_id": reviewer_id}

async def main():
    client = await Client.connect(os.environ.get("TEMPORAL_HOST", "localhost:7233"))
    async with Worker(client, task_queue="approval-queue", workflows=[DocumentApprovalWorkflow], activities=[run_ai_classifier, enqueue_human_review, record_final_decision]):
        handle = await client.start_workflow(DocumentApprovalWorkflow.run, "doc-001", id="approval-doc-001", task_queue="approval-queue")
        print("Started workflow:", handle.id)
        await asyncio.sleep(2)
        # Simulate reviewer submitting decision via signal
        await handle.signal(SIGNAL_HUMAN_DECISION, {"decision": "APPROVE", "reviewer_id": "reviewer-42"})
        result = await handle.result()
        print("Final:", result)

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