feat: add solution-first selling pipeline

.gitattributes

@@ -1,4 +1,6 @@
 # Hash-pinned Aliyun metadata and generated catalogs use canonical LF bytes.
 src/iac_code/tools/cloud/aliyun/data/** text eol=lf
 tests/tools/cloud/aliyun/fixtures/** text eol=lf
+# Pipeline definitions, prompts, and bundled skills are byte-validated release resources.
+src/iac_code/pipeline/selling_solution_first/** text eol=lf
 src/iac_code/web/static/js/vendor/mermaid.min.js -whitespace

babel.cfg

@@ -1,2 +1,2 @@
 [python: src/iac_code/**.py]
-keywords = translate_message
+keywords = translate_message CompletionEnrichmentError CompletionValidationError

scripts/a2a/e2e/README.md

@@ -13,6 +13,14 @@ copy can invalidate the source for the next scenario. It then fixes the server
 policy at `300 / 300 / 30`, enables the shared-backup commit protocol, uses
 unique Stack/VSwitch names, and performs an exact-name cleanup fallback.
 
+`--run-dir` remains the output root for logs and evidence and may use paths such
+as `/tmp/...`. The runner places the Qoder workspace and ROS Agent manager state
+under Python's `tempfile.gettempdir()` automatically, because the managed Skill
+accepts local manager paths only under the current user's home or
+Python-recognized temporary tree. This matters on macOS, where `/tmp` resolves
+to `/private/tmp` while Python commonly reports a different per-user temporary
+root.
+
 The Qoder flag that bypasses host Bash/file confirmation applies only to the
 test driver. It does not approve ROS Agent permissions: the isolated iac-code
 settings use the default permission mode, explicitly allow incidental tools,
@@ -99,6 +107,13 @@ credentials. The Sub-Pipeline fixture asserts one denial ToolResult, continued
 Agent-loop execution, parent candidate selection/completion, and the absence of
 grace, durable permission checkpoints, and permission-critical backup.
 
+The same restart suite also covers all three `selling_solution_first` steps
+(`solution_planning_and_selection`, `materialize_selected_candidate`, and
+`deploying`) plus normal chat after the pipeline handoff. Every scope runs both
+`allow_once` and `deny`, emits exactly one permission, restarts the real HTTP
+A2A server before answering, verifies the safe operation/parameter projection,
+and uses deterministic local tools only—no cloud write is performed.
+
 This directory contains headless end-to-end checks for A2A pipeline session
 recovery and redaction regressions. The runner drives the public A2A JSON-RPC
 streaming endpoint and records SSE events and pipeline snapshots. Recovery

scripts/a2a/e2e/README.zh-CN.md

@@ -9,6 +9,11 @@
 一次性副本中刷新并轮换 OAuth refresh token，导致源配置在下一个场景失效。随后把服务端策略固定为
 `300 / 300 / 30`，启用共享备份提交协议，使用唯一的 Stack/VSwitch 名称，并在结束时仅按精确名称做兜底清理。
 
+`--run-dir` 仍是日志和证据的输出根目录，可以继续使用 `/tmp/...` 等路径。runner 会自动把 Qoder 工作区和
+ROS Agent manager 状态放到 Python 的 `tempfile.gettempdir()` 下，因为托管 Skill 只接受位于当前用户主目录
+或 Python 所识别临时目录中的本地 manager 路径。这一点在 macOS 上尤其重要：`/tmp` 会解析为
+`/private/tmp`，而 Python 通常返回另一个用户级临时目录。
+
 Qoder 的 host 权限绕过只用于允许测试驱动执行本地 Bash/文件操作，不会批准 ROS Agent 权限。隔离的 iac-code
 配置使用默认权限模式，显式允许辅助工具并要求云资源变更工具确认；A2A server 同时保持
 `auto_approve_permissions: false`，非只读云操作仍必须通过带完整关联字段的 StartChat 权限回答。
@@ -80,6 +85,12 @@ uv run pytest -q tests/a2a_e2e/test_permission_wait_restart.py
 Agent loop 继续、父 Pipeline 进入 candidate 选择并完成，且全程没有 grace、持久化 permission checkpoint
 或权限关键备份。
 
+同一套重启测试还覆盖 `selling_solution_first` 的三个步骤
+（`solution_planning_and_selection`、`materialize_selected_candidate`、`deploying`）以及
+Pipeline handoff 后的普通对话。每个作用域都分别执行 `allow_once` 和 `deny`，严格只触发一次权限，
+在回答前重启真实 HTTP A2A server，并校验安全 operation/参数投影。fixture 只执行确定性的本地工具，
+不会产生真实云写操作。
+
 本目录包含用于 A2A pipeline 会话恢复和脱敏回归的 headless 端到端检查。Runner 会驱动公开的
 A2A JSON-RPC streaming endpoint 并记录 SSE 事件和 pipeline snapshot。恢复场景会用 `SIGKILL`
 杀掉 A2A server，再用相同持久化目录重启；`redaction-step4` 则在候选方案选择处停止，不重启、

scripts/a2a/e2e/fixtures/backup-delay-sitecustomize/sitecustomize.py

@@ -16,7 +16,7 @@
 _CONTROL_ENV = "IAC_CODE_E2E_BACKUP_DELAY_CONTROL"
 _ARM_WAIT_SECONDS = 5.0
 _claim_lock = threading.Lock()
-_claimed = False
+_claimed_controls: set[Path] = set()
 
 
 def _marker_path(control: Path, marker: str) -> Path:
@@ -31,8 +31,6 @@ def _write_marker(path: Path, payload: dict[str, Any]) -> None:
 
 
 def _claim_delay(reason: Any) -> tuple[Path, float, float] | None:
-    global _claimed
-
     reason_value = getattr(reason, "value", reason)
     if reason_value != session_backup.BackupReason.INPUT_REQUIRED.value:
         return None
@@ -44,17 +42,25 @@ def _claim_delay(reason: Any) -> tuple[Path, float, float] | None:
     if not control_value or delay_seconds <= 0:
         return None
 
-    control = Path(control_value)
-    arm_path = _marker_path(control, "arm")
-    with _claim_lock:
-        if _claimed:
-            return None
-        deadline = time.monotonic() + _ARM_WAIT_SECONDS
-        while not arm_path.is_file() and time.monotonic() < deadline:
+    configured_control = Path(control_value)
+    deadline = time.monotonic() + _ARM_WAIT_SECONDS
+    control: Path | None = None
+    while control is None and time.monotonic() < deadline:
+        if configured_control.is_dir():
+            candidates = sorted(
+                arm_path.with_name(arm_path.name.removesuffix(".arm.json"))
+                for arm_path in configured_control.glob("backup-delay-*.arm.json")
+            )
+        else:
+            candidates = [configured_control] if _marker_path(configured_control, "arm").is_file() else []
+        with _claim_lock:
+            control = next((candidate for candidate in candidates if candidate not in _claimed_controls), None)
+            if control is not None:
+                _claimed_controls.add(control)
+        if control is None:
             time.sleep(0.02)
-        if not arm_path.is_file():
-            return None
-        _claimed = True
+    if control is None:
+        return None
 
     started_at = time.time()
     started_monotonic = time.monotonic()