cxas-dfcx-migration

DFCX 到 CXAS 迁移

四个小脚本，一个持久化的 IR 包：

脚本	功能	运行时间	输出
migrate.py	将每个选定的 playbook/flow 逐一转换为 IR，并且仅部署基础资源（应用、变量、工具）。Agent 部署延后至 stage_1.py，以免大型源超出 CXAS 的 100 个 Agent 上限——编译后的 Agent 会保存到 <target>_ir.json，不会推送。传入 --no-consolidate 可立即推送完整的 1:1 Agent 集（仅在 Agent 数量低于约 100 时安全）。	约 40 个 flow 需 30 分钟	<target>_ir.json、<target>_migration_report.md、<target>_unit_tests.json
stage_1.py	加载 IR 包，运行 CXASOptimizer.optimize_stage1（变量去重）和 Gemini 结构合并（将 N 个 Agent 分组为 M 个）。这是首次向 CXAS 推送 Agent——仅部署合并后的（N→M）Agent；不会推送原始的 1:1 Agent（consolidate() 会丢弃这些 Agent，而合并前的快照仅临时用于完整性检查，不会持久化）。CXAS 版本 0.0.2（去重）和 0.0.3（合并）。	约 15 分钟	更新后的 <target>_ir.json、<target>_grouping.json
stage_2.py	加载 IR 包，运行 CXASOptimizer.optimize_stage2（指令状态机 + 工具模拟）。通过更新式部署推送。CXAS 版本 0.0.4。重新生成单元测试。执行 lint。写入审计报告。	约 10 分钟	更新后的 <target>_ir.json、<target>_optimization_report.md、重新生成的 <target>_unit_tests.json
stage_3.py	仅在 Stage 1 合并完成后运行。 将 SOURCE DFCX 依赖图映射到新分组（而不是依赖合成的 PIF XML 中碰巧引用的内容），按照 Spoke-Hub 架构风格，重新连接合并后 Agent 的父子拓扑。将应用的 root_agent 设为 is_root 分组。幂等，可安全重复运行。CXAS 版本 0.0.5。	约 10 秒	更新后的 <target>_ir.json 阶段历史；CXAS 应用中每个分组的 child_agents

状态通过 <target>_ir.json 流转（一个 Pydantic IRBundle，其中包含 MigrationConfig、源 DFCXAgentIR、目标 MigrationIR、阶段历史和版本检查点）。每个阶段都会从磁盘加载它，修改后再写回。阶段之间不会重新获取或重新编译。

何时使用此技能，何时直接使用 CLI

此技能（InquirerPy 提示 + HTML 预检预览 + Gemini 模型选择器）适合在需要交互式执行迁移并查看丰富预检上下文时使用。此技能调用的同一组 MigrationService.run_stage* 方法也通过规范 CLI 提供，适用于脚本或 CI：

# Same E2E plumbing, non-interactive (standard optimized profile by default):
cxas migrate dfcx --run --source-agent-id … --project-id … --target-name …

# Non-interactive Stage Checkpoint optimization runs:
cxas migrate dfcx --optimize --stage 1 --target-name my_app
cxas migrate dfcx --optimize --stage 2 --target-name my_app
cxas migrate dfcx --optimize --stage 3 --target-name my_app --architecture hub-and-spoke
cxas migrate dfcx --optimize --stage resume --target-name my_app  # interactive stage picker

此技能、仪表板以及 E2E / Checkpoint 命令都会调用同一组 MigrationService.run_stage_1/run_stage_2/run_stage_3 方法——请选择符合你工作流的入口。

前提条件

# Ensure cxas_scrapi is installed editable (so this skill picks up local changes)
pip install -e .

# Auth
gcloud auth application-default login
gcloud auth list   # confirm the account has read on source + admin on target

交互式提示需要 InquirerPy（与 agent-foundry 技能一致）：

pip install InquirerPy

交互式执行流程（通过 Claude）

通过 Claude 调用时，每次只向用户提出一个问题。脚本会通过 InquirerPy 提示输入缺失项，但你应预先收集：

1. GCP 项目 ID（目标——新 CXAS 应用所在的位置）。
2. 位置——默认值为 us。不要将 global 设为默认值——在大多数项目中，它无法用于 CXAS 应用。
3. 源 Agent——DFCX Agent ID（完整资源名称）或本地 .zip 导出文件的路径。
4. 目标名称——新 CXAS 应用的显示名称。

可选后续参数：--env（PROD/AUTOPUSH）、--model（Gemini）、--migration-version（1.0/2.0）。

快速参考

# 1:1 migration (interactive — InquirerPy will prompt for project + location)
python .agents/skills/cxas-dfcx-migration/scripts/migrate.py

# Fully scripted
python .agents/skills/cxas-dfcx-migration/scripts/migrate.py \
  --source-agent-id "projects/<src_proj>/locations/us/agents/<uuid>" \
  --project-id <target_proj> --location us \
  --target-name my_cxas_app --yes

# Pre-flight HTML preview only (no migration)
python .agents/skills/cxas-dfcx-migration/scripts/migrate.py \
  --source-agent-id "<id>" --project-id <proj> --target-name preview_only \
  --preview-only --yes

# Stage 1 — variable dedup + Gemini consolidation
python .agents/skills/cxas-dfcx-migration/scripts/stage_1.py --target-name my_cxas_app

# Stage 1 — replay a saved grouping JSON
python .agents/skills/cxas-dfcx-migration/scripts/stage_1.py \
  --target-name my_cxas_app --grouping-json my_cxas_app_grouping.json --yes

# Stage 2 — instruction state machines + tool mocks + lint + report
python .agents/skills/cxas-dfcx-migration/scripts/stage_2.py --target-name my_cxas_app

# Stage 3 — rewire consolidated agent parent-child topology (idempotent)
python .agents/skills/cxas-dfcx-migration/scripts/stage_3.py --target-name my_cxas_app --architecture hub-and-spoke

技能中与 cxas_scrapi 中的内容

技能只是一个轻量级编排器。所有迁移／优化步骤都位于 src/cxas_scrapi/migration/ 中，并通过 MigrationService.run_stage_* 方法调用，因此三个入口都使用相同的逻辑：此技能、cxas migrate dfcx（交互式 TUI）以及非交互式命令模式（--run／--optimize）。

操作	src/ 入口点
获取源 Agent／解析 zip	migration/dfcx_exporter.py:ConversationalAgentsAPI
1:1 迁移	migration/service.py:MigrationService.run_migration
阶段 1 编排器（变量去重 + 整合 + 完整性检查 + 拓扑链接 + 孤立项清理 + 版本 + 保存捆绑包）	migration/service.py:MigrationService.run_stage_1
阶段 2 编排器（状态机 + 工具模拟 + 重新生成单元测试 + lint + 审计报告 + 保存捆绑包）	migration/service.py:MigrationService.run_stage_2
阶段 3 编排器（父子拓扑连接）	migration/service.py:MigrationService.run_stage_3
保存捆绑包的便捷方法	migration/service.py:MigrationService.persist_bundle
变量去重基础操作（阶段 1）	migration/optimizer.py:CXASOptimizer.optimize_stage1
指令重组 + 工具模拟基础操作（阶段 2）	migration/optimizer.py:CXASOptimizer.optimize_stage2
Gemini N→M 分组 + 按组生成 PIF XML	migration/structural_consolidator.py:StructuralConsolidator
部署前完整性检查	migration/integrity_checks.py:check_consolidation_integrity
父子拓扑 + 孤立项清理	migration/topology_wirer.py
更新轮次重新部署	migration/service.py:MigrationService._deploy_base_resources(is_update_pass=True) + _deploy_pending_agents(is_update_pass=True)
拓扑链接	migration/cxas_topology_linker.py
版本检查点	core/versions.py:Versions.create_version
拓扑 SVG	migration/graph_visualizer.py:HighLevelGraphVisualizer
每个资源的 Rich 树	migration/playbook_visualizer.py + migration/flow_visualizer.py
确定性单元测试	migration/eval_generator.py:DeterministicEvalGenerator
迁移报告	migration/dfcx_migration_reporter.py
优化审计报告	migration/optimization_reporter.py:OptimizationReporter
分组审核 TUI（接受／重新提议／合并／拆分／重命名）	cli/grouping_review.py:interactive_review
HTML 部署前预览	migration/html_preview.py
部署后 lint	migration/post_deploy_lint.py
IR 捆绑包持久化	migration/data_models.py:IRBundle

技能本地辅助工具（仅用于 UX 粘合层——InquirerPy 提示词 + 对 MigrationCLI 的轻量委派）：

• _prompts.py — InquirerPy 提示词库（与 agent-foundry 保持一致）。
• _shared.py — InquirerPy 版项目／位置提示词、源加载器和 MigrationConfig 组装；另有对 MigrationCLI 的纯委派，用于 check_auth、run_dependency_analysis、select_resources、show_visualizations。

阶段脚本现在直接导入提升后的模块： from cxas_scrapi.migration.data_models import IRBundle （以及 migrate.py 中的 html_preview 和 phase_tracker），调用处使用规范模型名称（IRBundle、phase_tracker.PhaseTracker、 html_preview.generate_html_report）——不再有重新导出适配层。

技能的阶段脚本（migrate.py／stage_1.py／stage_2.py／stage_3.py）现在都是约 80–200 行的外壳：解析参数 → 从捆绑包恢复服务 → 调用对应的 MigrationService.run_stage_* → 打印摘要。技能中已没有编排逻辑——只剩 InquirerPy 提示词，以及技能部署前 UX 专用的 HTML 预览。技能部署前 UX 专用的 HTML 预览。

IR 包（<target>_ir.json）

在三个脚本之间共享状态的单元。Pydantic IRBundle 模型：

{
  "schema_version": "1",
  "created_at": "2026-05-14T15:30:00",
  "config": { /* MigrationConfig */ },
  "source_agent_data": { /* DFCXAgentIR — needed for tool-mock context */ },
  "ir": { /* MigrationIR — mutated by each stage */ },
  "stage_history": [
    {"phase": "migrate", "status": "ok", ...},
    {"phase": "stage1",  "status": "ok", ...}
  ],
  "app_url": "https://ces.cloud.google.com/...",
  "version_checkpoints": [["0.0.1", "Stage 1: ..."]],
  "grouping": { /* present if Stage 1 ran consolidation */ }
}

如果在运行中终止某个阶段脚本，IR 包不会受到影响（仅在成功后持久化）。重新运行会从上一个成功阶段结束的位置继续。

迁移前 HTML 预览

加载源数据后，migrate.py 会在约 5 秒内生成 <target>_tree_preview.html。在任意浏览器中打开即可查看：

• 源数据概览（资源数量、预计迁移时间）。
• 拓扑图（如果 dot 位于 PATH 中，则使用 graphviz SVG；否则使用 Mermaid）。
• 每个 playbook 和 flow 对应的 Rich 树。

migrate.py --preview-only 会在生成预览后退出，不会运行迁移。

故障排查

• create_app 返回 404 / 501 / MethodNotImplemented — --location 设置错误。大多数项目中的 CXAS 应用位于 us，而非 global。请传入 --location us。
• AlreadyExists: App with same display name — 请选择不同的 --target-name。即使运行部分失败，旧运行也会留下已部署的应用。
• 阶段 1 / 阶段 2 报错 No IR bundle found — 先运行 migrate.py 以生成 <target>_ir.json，或显式传入 --ir-bundle <path>。
• Gemini 上的合成（阶段 1 整合）卡住 — 已在 structural_consolidator.synthesize_instructions 中通过每组 asyncio.wait_for(timeout=600s) 修复。可通过 SYNTHESIS_TIMEOUT_S 环境变量覆盖。卡住的组会回退到拼接后的指令。
• AI 增补期间出现 gemini-2.5-flash-001 not found — Gemini 调用使用 locations/global 作为模型位置；该端点有时会受到项目限制。迁移会继续进行，但 AI 描述为空。如有需要，请选择其他 --model。

详细参考

完整的参数 / 标志说明以及 IR 包架构，请参阅 references/migration-options.md。