feat(ai): temporal worker for lesson plan workflow

- deploy temporal server (postgresql + auto-setup + ui) in docker-compose - new temporal/ module: workflow + activities + worker manager - convert lesson plan 4-step orchestration to temporal workflow - activities wrap existing analyze/recommend/generate/prepare_review steps - worker injects failover_chain/content_client/data_ana_client via module globals - start() uses temporal client.start_workflow, falls back to asyncio in dev - register temporal ports 7233/8085 in port-allocation Implements M6.5 of v2.1 migration plan (ADR-030).
2026-07-15 02:34:34 +08:00
parent ce5aeec955
commit 47e950c664
11 changed files with 701 additions and 40 deletions
--- a/infra/docker-compose.yml
+++ b/infra/docker-compose.yml
@@ -346,6 +346,63 @@ services:
      interval: 15s
      timeout: 5s
      retries: 5
+  # ============================================================
+  # Temporal Server - AI 工作流引擎（v2.1 §8.2 ADR-030）
+  # 仅用于 AI 耗时工作流 + Saga，CRUD 短事务禁止
+  # ============================================================
+  temporal-postgresql:
+    image: docker.m.daocloud.io/library/postgres:13
+    container_name: edu-temporal-postgres
+    profiles: ["p3", "p4", "p5", "p6"]
+    restart: unless-stopped
+    environment:
+      POSTGRES_USER: temporal
+      POSTGRES_PASSWORD: ${TEMPORAL_POSTGRES_PASSWORD:-temporal}
+      POSTGRES_DB: temporal
+    volumes:
+      - temporal_pg_data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U temporal"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+  temporal:
+    image: temporalio/auto-setup:1.23
+    container_name: edu-temporal
+    profiles: ["p3", "p4", "p5", "p6"]
+    restart: unless-stopped
+    depends_on:
+      temporal-postgresql:
+        condition: service_healthy
+    environment:
+      DBHOST: temporal-postgresql
+      DBPORT: 5432
+      DBUSER: temporal
+      DBPASSWORD: ${TEMPORAL_POSTGRES_PASSWORD:-temporal}
+      DBNAME: temporal
+      DB_PLUGIN: postgres
+      # 暴露 frontend gRPC 端口 7233 供 Worker 连接
+      SERVICES: "frontend,history,matching,worker"
+    ports:
+      - "7233:7233"
+    healthcheck:
+      test: ["CMD", "tctl", "--address", "localhost:7233", "cluster", "health"]
+      interval: 15s
+      timeout: 5s
+      start_period: 30s
+      retries: 10
+  temporal-ui:
+    image: temporalio/ui:2.30.0
+    container_name: edu-temporal-ui
+    profiles: ["p3", "p4", "p5", "p6"]
+    restart: unless-stopped
+    depends_on:
+      - temporal
+    environment:
+      TEMPORAL_ADDRESS: temporal:7233
+      TEMPORAL_CORS_ORIGINS: "http://localhost:4000,http://localhost:4001,http://localhost:4002,http://localhost:4003"
+    ports:
+      - "8085:8080"
 volumes:
  mysql_data:
  redis_data:
@@ -356,3 +413,4 @@ volumes:
  prometheus_data:
  alertmanager_data:
  loki_data:
+  temporal_pg_data:
--- a/infra/port-allocation.md
+++ b/infra/port-allocation.md
@@ -94,7 +94,7 @@
 ## 6. 基础设施端口（9080-9199）

 | 服务              | 端口                                                                        | 说明                                  |
-| ---------------- | --------------------------------------------------------------------------- | ------------------------------------- |
+| ----------------- | --------------------------------------------------------------------------- | ------------------------------------- |
 | MySQL             | 3306                                                                        | 业务数据库（external）                |
 | Redis             | 6379                                                                        | 缓存 + 会话 + Pub/Sub                 |
 | Kafka             | 9092（OUTSIDE）/ 29092（INSIDE）                                            | 双监听器                              |
@@ -111,6 +111,9 @@
 | node-exporter     | 9100                                                                        | 主机指标（host.docker.internal:9100） |
 | mysqld-exporter   | 9104                                                                        | MySQL 指标                            |
 | redis-exporter    | 9121                                                                        | Redis 指标                            |
+| Temporal          | 7233                                                                        | gRPC frontend（Worker 连接）          |
+| Temporal UI       | 8085                                                                        | Temporal Web UI（容器内 8080）        |
+| Temporal Postgres | 5433（预留，内部 5432）                                                     | Temporal 持久化存储（仅容器内）       |

 > **Grafana 端口冲突风险**：Grafana 默认 3000，与 teacher-bff 3003 不冲突（不同主机层），但开发环境若同主机部署需注意。建议 Grafana 改用 3030 避免混淆。

@@ -119,12 +122,13 @@
 ## 7. 变更记录

 | 日期       | 变更                                                              | 决策者 |
-| ---------- | --------------------------------------------------------- | ------ |
+| ---------- | ----------------------------------------------------------------- | ------ |
 | 2026-07-09 | 初始创建，登记全部 15 服务 + 基础设施端口                         | coord  |
 | 2026-07-09 | 仲裁 admin-portal 3003 → 4003；MF 配置 3000 → 4000                | coord  |
 | 2026-07-09 | 仲裁 push-gateway 豁免 gRPC，释放 50057；50058 让给 ai            | coord  |
 | 2026-07-09 | classes 3001 标记为历史（已合并入 core-edu）                      | coord  |
 | 2026-07-14 | M3 新增 config-service（3011/50059，ADR-026 从 iam 拆分）         | coord  |
+| 2026-07-14 | M6.5 新增 Temporal（7233/UI 8085/PG 5433，ADR-030 AI 工作流引擎） | coord  |

 ---

--- a/infra/security/secrets.example.env
+++ b/infra/security/secrets.example.env
@@ -55,3 +55,8 @@ ENCRYPTION_KEY=<replace-with-base64-32-byte-aes-key>
 # 生成：openssl rand -hex 32
 # 注意：Router 和所有子图必须使用相同的密钥
 ROUTER_AUTH_SECRET=<replace-with-32-char-router-auth-secret>
+
+# ---------- Temporal ----------
+# 用途：Temporal PostgreSQL 存储密码
+# 最小长度：24 字符
+TEMPORAL_POSTGRES_PASSWORD=<replace-with-24-char-temporal-password>
--- a/services/ai/pyproject.toml
+++ b/services/ai/pyproject.toml
@@ -22,6 +22,8 @@ dependencies = [
    "tenacity>=9.0.0",
    # GraphQL Federation 2 子图（v2.1 M1，Apollo Router 组合）
    "strawberry-graphql[asgi]>=0.257.0",
+    # Temporal 工作流引擎（v2.1 §8.2 ADR-030，AI 耗时工作流）
+    "temporalio>=1.7.0",
 ]

 [tool.uv.sources]
--- a/services/ai/src/ai/config.py
+++ b/services/ai/src/ai/config.py
@@ -64,6 +64,13 @@ class Settings(BaseSettings):
    workflow_ttl_seconds: int = 3600  # 1h
    workflow_max_retries: int = 3

+    # Temporal（v2.1 §8.2：AI 耗时工作流引擎）
+    temporal_host: str = "localhost:7233"
+    temporal_namespace: str = "default"
+    temporal_task_queue: str = "ai-lesson-plan"
+    temporal_workflow_timeout_seconds: int = 3600  # 1h
+    temporal_activity_timeout_seconds: int = 300  # 5min per activity
+
    # 评估
    evaluation_pass_threshold: float = 0.7
    evaluation_excellent_threshold: float = 0.85
--- a/services/ai/src/ai/main.py
+++ b/services/ai/src/ai/main.py
@@ -74,6 +74,7 @@ from .providers import create_failover_chain
 from .rate_limiter import RateLimiter
 from .services import ChatService, ExpressionService, QuestionService, ReportService
 from .services.evaluation import QualityGate, RuleValidator
+from .temporal import TemporalWorkerManager
 from .usage import KafkaProducer, QuotaEnforcer, UsageRecorder
 from .workflow import LessonPlanWorkflowService, WorkflowStateStore

@@ -178,6 +179,7 @@ _grpc_server = create_grpc_server(
 )

 _redis: Redis | None = None
+_temporal_worker: TemporalWorkerManager | None = None


@asynccontextmanager
@@ -215,6 +217,32 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
                error=str(exc),
            )

+    # Temporal Worker（v2.1 M6.5 ADR-030：备课工作流引擎）
+    # 连接失败时降级（LessonPlanWorkflowService 自动回退 asyncio），不阻断启动
+    global _temporal_worker
+    try:
+        _temporal_worker = TemporalWorkerManager(
+            host=settings.temporal_host,
+            namespace=settings.temporal_namespace,
+            task_queue=settings.temporal_task_queue,
+        )
+        temporal_client = await _temporal_worker.start(
+            failover_chain=_failover_chain,
+            prompt_service=_prompt_service,
+            quality_gate=_quality_gate,
+            content_client=_content_client,
+            data_ana_client=_data_ana_client,
+            state_store=_state_store,
+            default_model=settings.default_question_model,
+        )
+        _workflow_service.set_temporal_client(
+            temporal_client,
+            settings.temporal_task_queue,
+        )
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("temporal_worker_start_failed_degraded", error=str(exc))
+        _temporal_worker = None
+
    await _kafka_producer.start()
    await _grpc_server.start()

@@ -237,6 +265,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    logger.info("ai_service_stopping")
    await _grpc_server.stop()
    await _kafka_producer.stop()
+    if _temporal_worker is not None:
+        try:
+            await _temporal_worker.stop()
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("temporal_worker_stop_failed", error=str(exc))
    for name, client in reversed(downstream_clients):
        try:
            await client.close()
--- a/services/ai/src/ai/temporal/init.py
+++ b/services/ai/src/ai/temporal/init.py
@@ -0,0 +1,14 @@
+"""Temporal 工作流引擎模块（v2.1 §8.2 ADR-030）.
+
+ai 服务作为 Temporal Worker，承载备课工作流（4 步编排）。
+边界：仅 AI 耗时工作流 + Saga，CRUD 短事务禁止走 Temporal。
+
+子模块：
+  - workflow: Temporal Workflow 定义（LessonPlanWorkflow）
+  - activities: 4 步编排 Activities
+  - worker: Worker 生命周期管理（TemporalWorkerManager）
+"""
+
+from .worker import TemporalWorkerManager
+
+__all__ = ["TemporalWorkerManager"]
--- a/services/ai/src/ai/temporal/activities.py
+++ b/services/ai/src/ai/temporal/activities.py
@@ -0,0 +1,365 @@
+"""Temporal Activities - 备课工作流 4 步编排（v2.1 §8.2 ADR-030）.
+
+将 lesson_plan_workflow.py 的 4 个 step 方法提取为 Temporal Activities。
+每个 Activity 封装一步业务逻辑，接收/返回可序列化的 dict。
+
+Activity 通过模块级 _clients 字典访问依赖（由 worker.py 启动时注入）：
+  - data_ana_client: 学情查询
+  - content_client: 知识点查询
+  - failover_chain: LLM 调用
+  - prompt_service: Prompt 模板渲染
+  - quality_gate: 题目评估三道防线
+  - state_store: Redis 工作流状态存储
+  - default_model: 默认 LLM 模型
+"""
+
+from typing import Any
+
+import structlog
+from temporalio import activity
+
+from ..clients.content_client import ContentClient
+from ..clients.data_ana_client import DataAnaClient
+from ..errors import AIError
+from ..models.question import GeneratedQuestionData
+from ..prompt_service import PromptTemplateService
+from ..providers import ProviderFailoverChain
+from ..services.evaluation import QualityGate
+from ..workflow.state_store import WorkflowState, WorkflowStateStore
+
+logger = structlog.get_logger()
+
+# 状态常量（与 lesson_plan_workflow.py 保持一致）
+STATUS_ANALYZING = "analyzing"
+STATUS_GENERATING = "generating"
+STATUS_PENDING_REVIEW = "pending_review"
+
+# 生成重试上限
+MAX_GENERATE_RETRIES = 3
+
+# 模块级依赖注册表（由 TemporalWorkerManager.start 注入）
+_clients: dict[str, Any] = {}
+
+
+def set_clients(**clients: Any) -> None:
+    """注入依赖到 Activities（由 worker 启动时调用）."""
+    _clients.update(clients)
+
+
+def _get_state_store() -> WorkflowStateStore:
+    """获取已注入的状态存储."""
+    store = _clients.get("state_store")
+    if store is None:
+        raise AIError("state_store not injected to temporal activities")
+    return store  # type: ignore[return-value]
+
+
+@activity.defn
+async def analyze_activity(state_data: dict) -> dict:
+    """Step 1: 分析学情.
+
+    调 data-ana 查询班级学情 + 学生薄弱点。
+    data-ana 不可用时降级（返回空分析）。
+    同时更新 Redis 状态：status=analyzing + analysis。
+    """
+    state = WorkflowState.from_dict(state_data)
+    store = _get_state_store()
+
+    await store.update(state.workflow_id, status=STATUS_ANALYZING)
+
+    data_ana_client: DataAnaClient | None = _clients.get("data_ana_client")
+    analysis: dict[str, Any] = {}
+
+    if data_ana_client is not None:
+        try:
+            performance = await data_ana_client.get_class_performance(
+                class_id=state.class_id,
+                subject_id=state.subject_id,
+            )
+            analysis["class_performance"] = {
+                "average_score": performance.average_score,
+                "pass_rate": performance.pass_rate,
+                "student_count": len(performance.scores),
+            }
+            analysis["weak_students"] = [
+                {"student_id": s.student_id, "score": s.score}
+                for s in performance.scores
+                if s.score < 60
+            ]
+        except Exception as exc:  # noqa: BLE001
+            logger.warning(
+                "activity_step1_analyze_degraded",
+                workflow_id=state.workflow_id,
+                error=str(exc),
+            )
+            analysis["degraded"] = True
+            analysis["degraded_reason"] = f"data-ana unavailable: {exc}"
+    else:
+        analysis["degraded"] = True
+        analysis["degraded_reason"] = "data-ana client not configured"
+
+    await store.update(state.workflow_id, analysis=analysis)
+
+    logger.info(
+        "activity_step1_completed",
+        workflow_id=state.workflow_id,
+        degraded=analysis.get("degraded", False),
+    )
+    return analysis
+
+
+@activity.defn
+async def recommend_activity(state_data: dict) -> list[dict]:
+    """Step 2: 推荐知识点.
+
+    调 content 查询学习路径。content 不可用时降级（基于 topic 推导）。
+    """
+    state = WorkflowState.from_dict(state_data)
+    content_client: ContentClient | None = _clients.get("content_client")
+    knowledge_points: list[dict[str, str]] = []
+
+    if content_client is not None:
+        try:
+            learning_path = await content_client.get_learning_path(
+                student_id=state.user_id,
+                subject_id=state.subject_id,
+            )
+            knowledge_points = [{"id": kp.id, "title": kp.title} for kp in learning_path]
+        except Exception as exc:  # noqa: BLE001
+            logger.warning(
+                "activity_step2_recommend_degraded",
+                workflow_id=state.workflow_id,
+                error=str(exc),
+            )
+
+    # 降级：基于 topic 推导知识点
+    if not knowledge_points:
+        knowledge_points = [
+            {"id": "kp_default_1", "title": f"{state.topic} - 基础概念"},
+            {"id": "kp_default_2", "title": f"{state.topic} - 进阶应用"},
+            {"id": "kp_default_3", "title": f"{state.topic} - 综合题"},
+        ]
+
+    logger.info(
+        "activity_step2_completed",
+        workflow_id=state.workflow_id,
+        knowledge_point_count=len(knowledge_points),
+    )
+    return knowledge_points
+
+
+@activity.defn
+async def generate_activity(
+    state_data: dict,
+    knowledge_points: list[dict],
+) -> list[dict]:
+    """Step 3: 生成题目.
+
+    使用 LLM 生成题目 + 评估三道防线。
+    评估未通过时重试（最多 MAX_GENERATE_RETRIES 次）。
+    返回 list[dict]（GeneratedQuestionData.model_dump()）。
+    """
+    state = WorkflowState.from_dict(state_data)
+    failover_chain: ProviderFailoverChain | None = _clients.get("failover_chain")
+    prompt_service: PromptTemplateService | None = _clients.get("prompt_service")
+    quality_gate: QualityGate | None = _clients.get("quality_gate")
+    default_model: str = _clients.get("default_model", "gpt-4o-mini")
+
+    kp_ids = [kp["id"] for kp in knowledge_points]
+
+    if failover_chain is None or quality_gate is None:
+        logger.warning(
+            "activity_step3_degraded_no_llm",
+            workflow_id=state.workflow_id,
+        )
+        return [
+            GeneratedQuestionData(
+                question=f"[degraded] LLM 未配置，请手动添加题目：{state.topic}",
+                answer="",
+                explanation="LLM provider 或 quality_gate 未注入",
+                question_type="short_answer",
+                difficulty=state.target_difficulty,
+                knowledge_point_ids=kp_ids,
+                evaluation_score=0.0,
+                degraded=True,
+                degraded_reason="failover_chain or quality_gate not injected",
+            ).model_dump(),
+        ]
+
+    questions: list[GeneratedQuestionData] = []
+    for i in range(state.question_count):
+        question = await _generate_single_question(
+            state=state,
+            kp_ids=kp_ids,
+            question_index=i,
+            failover_chain=failover_chain,
+            prompt_service=prompt_service,
+            quality_gate=quality_gate,
+            default_model=default_model,
+        )
+        questions.append(question)
+
+    logger.info(
+        "activity_step3_completed",
+        workflow_id=state.workflow_id,
+        question_count=len(questions),
+    )
+    return [q.model_dump() for q in questions]
+
+
+async def _generate_single_question(
+    state: WorkflowState,
+    kp_ids: list[str],
+    question_index: int,
+    failover_chain: ProviderFailoverChain,
+    prompt_service: PromptTemplateService | None,
+    quality_gate: QualityGate,
+    default_model: str,
+) -> GeneratedQuestionData:
+    """生成单道题目（含重试）."""
+    prompt = _render_generate_prompt(state, kp_ids, question_index, prompt_service)
+    messages = [
+        {"role": "system", "content": "你是一个专业的教育题目生成助手。"},
+        {"role": "user", "content": prompt},
+    ]
+
+    for attempt in range(MAX_GENERATE_RETRIES):
+        try:
+            response = await failover_chain.chat(messages, default_model, 0.7)
+
+            evaluation = await quality_gate.evaluate(
+                llm_output=response.content,
+                expected_difficulty=state.target_difficulty,
+                expected_question_type="short_answer",
+                subject=state.subject_id,
+            )
+
+            parsed = evaluation.rule_result.parsed if evaluation.rule_result else None
+
+            if parsed and evaluation.passed:
+                return GeneratedQuestionData(
+                    question=str(parsed.get("question", "")),
+                    answer=str(parsed.get("answer", "")),
+                    explanation=str(parsed.get("explanation", "")),
+                    question_type=str(
+                        parsed.get("question_type", "short_answer"),
+                    ),
+                    difficulty=str(
+                        parsed.get("difficulty", state.target_difficulty),
+                    ),
+                    knowledge_point_ids=list(
+                        parsed.get("knowledge_point_ids", kp_ids),
+                    ),
+                    evaluation_score=evaluation.score,
+                )
+
+            logger.warning(
+                "activity_generate_retry",
+                workflow_id=state.workflow_id,
+                attempt=attempt + 1,
+                score=evaluation.score,
+            )
+
+        except Exception as exc:  # noqa: BLE001
+            logger.warning(
+                "activity_generate_error_retry",
+                workflow_id=state.workflow_id,
+                attempt=attempt + 1,
+                error=str(exc),
+            )
+
+    logger.warning(
+        "activity_generate_all_retries_failed",
+        workflow_id=state.workflow_id,
+        question_index=question_index,
+    )
+    return GeneratedQuestionData(
+        question=f"[degraded] 生成失败，请手动添加题目：{state.topic}",
+        answer="",
+        explanation="题目生成失败，已达到最大重试次数",
+        question_type="short_answer",
+        difficulty=state.target_difficulty,
+        knowledge_point_ids=kp_ids,
+        evaluation_score=0.0,
+        degraded=True,
+        degraded_reason="max retries exceeded",
+    )
+
+
+def _render_generate_prompt(
+    state: WorkflowState,
+    kp_ids: list[str],
+    question_index: int,
+    prompt_service: PromptTemplateService | None,
+) -> str:
+    """渲染题目生成 prompt."""
+    if prompt_service is not None:
+        try:
+            return prompt_service.render(
+                "lesson_plan_generate",
+                {
+                    "subject": state.subject_id,
+                    "topic": state.topic,
+                    "difficulty": state.target_difficulty,
+                    "knowledge_points": kp_ids,
+                    "knowledge_point_ids": kp_ids,
+                    "question_index": question_index,
+                    "analysis": state.analysis,
+                },
+            )
+        except Exception:  # noqa: BLE001
+            pass
+
+    # 降级 prompt
+    return (
+        f"请为 {state.subject_id} 学科生成一道题目。\n"
+        f"主题：{state.topic}\n"
+        f"难度：{state.target_difficulty}\n"
+        f"知识点：{', '.join(kp_ids)}\n"
+        "请按 JSON 格式输出："
+        '{"question":"...","answer":"...","explanation":"..."}'
+    )
+
+
+@activity.defn
+async def prepare_review_activity(
+    state_data: dict,
+    questions: list[dict],
+) -> dict:
+    """Step 4: 设置为待审核.
+
+    将生成的题目写入 Redis 状态，状态置为 pending_review。
+    """
+    state = WorkflowState.from_dict(state_data)
+    store = _get_state_store()
+
+    questions_objs = [GeneratedQuestionData(**q) for q in questions]
+    await store.update(
+        state.workflow_id,
+        status=STATUS_PENDING_REVIEW,
+        questions=questions_objs,
+    )
+
+    logger.info(
+        "activity_step4_completed",
+        workflow_id=state.workflow_id,
+        question_count=len(questions_objs),
+    )
+    return {
+        "workflow_id": state.workflow_id,
+        "status": STATUS_PENDING_REVIEW,
+        "question_count": len(questions_objs),
+    }
+
+
+@activity.defn
+async def update_status_activity(payload: dict) -> dict:
+    """更新 Redis 工作流状态（通用）.
+
+    payload: {"workflow_id": str, "status": str}
+    """
+    store = _get_state_store()
+    workflow_id = payload["workflow_id"]
+    status = payload["status"]
+    await store.update(workflow_id, status=status)
+    return {"workflow_id": workflow_id, "status": status}
--- a/services/ai/src/ai/temporal/worker.py
+++ b/services/ai/src/ai/temporal/worker.py
@@ -0,0 +1,80 @@
+"""Temporal Worker 管理器（v2.1 §8.2 ADR-030）.
+
+负责：
+  - 连接 Temporal Server
+  - 注入依赖到 Activities（failover_chain / content_client / data_ana_client 等）
+  - 启动 Worker（注册 Workflow + Activities）
+  - 优雅关闭
+
+ai 服务作为 Temporal Worker，task_queue=ai-lesson-plan。
+"""
+
+import asyncio
+from typing import Any
+
+import structlog
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from . import activities, workflow
+
+logger = structlog.get_logger()
+
+
+class TemporalWorkerManager:
+    """Temporal Worker 生命周期管理."""
+
+    def __init__(self, host: str, namespace: str, task_queue: str) -> None:
+        self._host = host
+        self._namespace = namespace
+        self._task_queue = task_queue
+        self._client: Client | None = None
+        self._worker: Worker | None = None
+
+    async def start(self, **clients: Any) -> Client:
+        """启动 Worker.
+
+        Args:
+            **clients: 注入到 Activities 的依赖
+                - failover_chain: ProviderFailoverChain
+                - prompt_service: PromptTemplateService
+                - quality_gate: QualityGate
+                - content_client: ContentClient
+                - data_ana_client: DataAnaClient
+                - state_store: WorkflowStateStore
+                - default_model: str
+
+        Returns:
+            Temporal Client（供 LessonPlanWorkflowService 启动 Workflow）
+        """
+        activities.set_clients(**clients)
+
+        self._client = await Client.connect(self._host, namespace=self._namespace)
+        self._worker = Worker(
+            self._client,
+            task_queue=self._task_queue,
+            workflows=[workflow.LessonPlanWorkflow],
+            activities=[
+                activities.analyze_activity,
+                activities.recommend_activity,
+                activities.generate_activity,
+                activities.prepare_review_activity,
+                activities.update_status_activity,
+            ],
+        )
+        asyncio.create_task(self._worker.run())
+        logger.info(
+            "temporal_worker_started",
+            host=self._host,
+            namespace=self._namespace,
+            task_queue=self._task_queue,
+        )
+        return self._client
+
+    async def stop(self) -> None:
+        """优雅关闭 Worker."""
+        if self._worker is not None:
+            await self._worker.shutdown()
+            logger.info("temporal_worker_stopped")
+        self._worker = None
+        self._client = None
--- a/services/ai/src/ai/temporal/workflow.py
+++ b/services/ai/src/ai/temporal/workflow.py
@@ -0,0 +1,69 @@
+"""Temporal Workflow - 备课工作流编排（v2.1 §8.2 ADR-030）.
+
+4 步编排（每步为独立 Activity，可自动重试 + 状态持久化）：
+  Step 1: 分析学情（analyze_activity）
+  Step 2: 推荐知识点（recommend_activity）
+  Step 3: 生成题目（update_status_activity → generate_activity）
+  Step 4: 设置待审核（prepare_review_activity）
+
+边界约束（spec §8.2）：仅 AI 耗时工作流走 Temporal，CRUD 短事务禁止。
+"""
+
+from datetime import timedelta
+
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from . import activities
+
+
+@workflow.defn
+class LessonPlanWorkflow:
+    """备课工作流（Temporal 编排）."""
+
+    @workflow.run
+    async def run(self, state_data: dict) -> dict:
+        """执行 4 步编排.
+
+        Args:
+            state_data: WorkflowState.to_dict() 序列化数据
+
+        Returns:
+            {"workflow_id", "status", "question_count"}
+        """
+        # Step 1: 分析学情
+        analysis = await workflow.execute_activity(
+            activities.analyze_activity,
+            state_data,
+            start_to_close_timeout=timedelta(seconds=300),
+        )
+        state_data["analysis"] = analysis
+
+        # Step 2: 推荐知识点
+        knowledge_points = await workflow.execute_activity(
+            activities.recommend_activity,
+            state_data,
+            start_to_close_timeout=timedelta(seconds=300),
+        )
+
+        # Step 3: 生成题目
+        await workflow.execute_activity(
+            activities.update_status_activity,
+            {"workflow_id": state_data["workflow_id"], "status": "generating"},
+            start_to_close_timeout=timedelta(seconds=10),
+        )
+        questions = await workflow.execute_activity(
+            activities.generate_activity,
+            state_data,
+            knowledge_points,
+            start_to_close_timeout=timedelta(seconds=600),
+        )
+
+        # Step 4: 设置待审核
+        result = await workflow.execute_activity(
+            activities.prepare_review_activity,
+            state_data,
+            questions,
+            start_to_close_timeout=timedelta(seconds=30),
+        )
+        return result
--- a/services/ai/src/ai/workflow/lesson_plan_workflow.py
+++ b/services/ai/src/ai/workflow/lesson_plan_workflow.py
@@ -10,6 +10,10 @@
 P5 实现：FastAPI BackgroundTasks + Redis 状态存储
 P6+ 评估：Temporal 工作流引擎

+v2.1 M6.5：迁移到 Temporal Workflow（ADR-030）。
+  - temporal_client 可用时：start_workflow 由 Temporal 编排
+  - temporal_client 不可用时：降级 asyncio.create_task（开发模式）
+
 4 步编排：
  Step 1: 分析学情（调 data-ana.GetClassPerformance + GetStudentWeakness）
  Step 2: 推荐知识点（调 content.GetPrerequisites + GetLearningPath）
@@ -18,7 +22,7 @@ P6+ 评估：Temporal 工作流引擎
 """

 import asyncio
-from typing import Any
+from typing import TYPE_CHECKING, Any

 import structlog

@@ -31,6 +35,9 @@ from ..providers import ProviderFailoverChain
 from ..services.evaluation import QualityGate
 from .state_store import WorkflowState, WorkflowStateStore

+if TYPE_CHECKING:
+    from temporalio.client import Client
+
 logger = structlog.get_logger()

 # 状态常量（简化 proto WorkflowStatus，内部用）
@@ -51,8 +58,7 @@ ESTIMATED_COMPLETION_SECONDS = 60
 class LessonPlanWorkflowService:
    """备课工作流服务.

-    P5 使用 asyncio.create_task 在后台执行工作流。
-    P6+ 评估迁移到 Temporal。
+    v2.1 M6.5：temporal_client 可用时由 Temporal 编排，否则降级 asyncio.create_task。
    """

    def __init__(
@@ -64,6 +70,8 @@ class LessonPlanWorkflowService:
        content_client: ContentClient | None = None,
        data_ana_client: DataAnaClient | None = None,
        default_model: str = "gpt-4o-mini",
+        temporal_client: "Client | None" = None,
+        task_queue: str = "ai-lesson-plan",
    ) -> None:
        self._store = state_store
        self._chain = failover_chain
@@ -73,6 +81,17 @@ class LessonPlanWorkflowService:
        self._data_ana_client = data_ana_client
        self._default_model = default_model
        self._background_tasks: dict[str, asyncio.Task[Any]] = {}
+        self._temporal_client = temporal_client
+        self._task_queue = task_queue
+
+    def set_temporal_client(
+        self,
+        client: "Client",
+        task_queue: str = "ai-lesson-plan",
+    ) -> None:
+        """注入 Temporal Client（lifespan 中由 main.py 调用）."""
+        self._temporal_client = client
+        self._task_queue = task_queue

    async def start(
        self,
@@ -115,7 +134,18 @@ class LessonPlanWorkflowService:
        )
        await self._store.create(state)

-        # 启动后台任务执行工作流
+        if self._temporal_client is not None:
+            # v2.1 M6.5: 使用 Temporal Workflow 编排（ADR-030）
+            from ..temporal.workflow import LessonPlanWorkflow
+
+            await self._temporal_client.start_workflow(
+                LessonPlanWorkflow.run,
+                args=[state.to_dict()],
+                id=f"lesson-plan-{state.workflow_id}",
+                task_queue=self._task_queue,
+            )
+        else:
+            # 降级：asyncio.create_task（开发模式或 Temporal 不可用）
            task = asyncio.create_task(self._run_workflow(state.workflow_id))
            self._background_tasks[state.workflow_id] = task

@@ -124,6 +154,7 @@ class LessonPlanWorkflowService:
            workflow_id=state.workflow_id,
            user_id=user_id,
            topic=topic,
+            engine="temporal" if self._temporal_client is not None else "asyncio",
        )
        return state

@@ -361,10 +392,7 @@ class LessonPlanWorkflowService:
                    student_id=state.user_id,
                    subject_id=state.subject_id,
                )
-                knowledge_points = [
-                    {"id": kp.id, "title": kp.title}
-                    for kp in learning_path
-                ]
+                knowledge_points = [{"id": kp.id, "title": kp.title} for kp in learning_path]
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "workflow_step2_recommend_degraded",
@@ -444,11 +472,7 @@ class LessonPlanWorkflowService:
                    subject=state.subject_id,
                )

-                parsed = (
-                    evaluation.rule_result.parsed
-                    if evaluation.rule_result
-                    else None
-                )
+                parsed = evaluation.rule_result.parsed if evaluation.rule_result else None

                if parsed and evaluation.passed:
                    return GeneratedQuestionData(