fix(a2a): 收敛取消/回滚/候选重跑后 Pipeline 快照的非终态悬挂节点

src/iac_code/a2a/executor.py

@@ -1731,9 +1731,7 @@ def _resolve_cwd(self, metadata: Any | None) -> str:
             raise ValueError("Invalid A2A workspace metadata.")
         logical_cwd = os.path.normpath(cwd)
         resolved_cwd = resolve_workspace_path(Path(logical_cwd))
-        if not trust_request_cwd() and not any(
-            _is_relative_to(resolved_cwd, root) for root in _allowed_cwd_roots()
-        ):
+        if not trust_request_cwd() and not any(_is_relative_to(resolved_cwd, root) for root in _allowed_cwd_roots()):
             raise ValueError("Invalid A2A workspace metadata.")
         if resolved_cwd.exists():
             if not resolved_cwd.is_dir():

src/iac_code/a2a/input_required.py

@@ -178,35 +178,29 @@ def permission_display_fields(request: PermissionRequestEvent, *, language: str
             command = safe_input.get("command") or safe_input.get("cmd")
         if isinstance(command, str) and command.strip():
             command_fallback = translate_message("shell command", language=language)
-            target = translate_message(
-                "the current local workspace; command: {command}", language=language
-            ).format(command=_display_text(command, fallback=command_fallback, maximum=240))
+            target = translate_message("the current local workspace; command: {command}", language=language).format(
+                command=_display_text(command, fallback=command_fallback, maximum=240)
+            )
         else:
             target = translate_message("the current local workspace", language=language)
         effect = "read" if is_read_only else ("local_execution" if read_only_known else "unknown")
     elif tool_name in {"write_file", "edit_file"}:
         title = translate_message("Change a workspace file", language=language)
-        purpose = translate_message(
-            "Write a file needed for the requested infrastructure task.", language=language
-        )
+        purpose = translate_message("Write a file needed for the requested infrastructure task.", language=language)
         target = _safe_input_target(safe_input, language=language) or translate_message(
             "a file in the current workspace", language=language
         )
         effect = "file_change"
     elif tool_name in {"read_file", "glob", "grep"} or is_read_only:
         title = translate_message("Read workspace data with {tool}", language=language).format(tool=public_tool)
-        purpose = translate_message(
-            "Read local data needed for the requested infrastructure task.", language=language
-        )
+        purpose = translate_message("Read local data needed for the requested infrastructure task.", language=language)
         target = _safe_input_target(safe_input, language=language) or translate_message(
             "the current local workspace", language=language
         )
         effect = "read"
     else:
         title = translate_message("Run {tool}", language=language).format(tool=public_tool)
-        purpose = translate_message(
-            "Run this operation for the requested infrastructure task.", language=language
-        )
+        purpose = translate_message("Run this operation for the requested infrastructure task.", language=language)
         target = _safe_input_target(safe_input, language=language) or translate_message(
             "the current task workspace or cloud account", language=language
         )
@@ -319,9 +313,7 @@ def _cloud_operation_title(product: str, action: str, *, is_read_only: bool, lan
     if action == "CreateStack":
         return translate_message("Create {product} stack", language=language).format(product=product_label)
     if action == "ContinueCreateStack":
-        return translate_message("Continue creating {product} stack", language=language).format(
-            product=product_label
-        )
+        return translate_message("Continue creating {product} stack", language=language).format(product=product_label)
     if action == "UpdateStack":
         return translate_message("Update {product} stack", language=language).format(product=product_label)
     if action == "DeleteStack":
@@ -344,9 +336,7 @@ def _safe_input_target(value: Any, *, language: str) -> str:
     for key in ("file_path", "filePath", "path", "region_id", "regionId", "resource_id", "resourceId"):
         candidate = value.get(key)
         if isinstance(candidate, str) and candidate.strip():
-            return _display_text(
-                candidate, fallback=translate_message("the current task scope", language=language)
-            )
+            return _display_text(candidate, fallback=translate_message("the current task scope", language=language))
     return ""
 
 

src/iac_code/a2a/pipeline_executor.py

@@ -1235,11 +1235,7 @@ def _create_pipeline(
             def permission_context_getter() -> Any:
                 return getattr(agent_loop, "_permission_context", None)
 
-        surface = (
-            A2A_RICH_CANDIDATE_SURFACE
-            if self._candidate_presentation == RICH_CANDIDATE_PRESENTATION
-            else "a2a"
-        )
+        surface = A2A_RICH_CANDIDATE_SURFACE if self._candidate_presentation == RICH_CANDIDATE_PRESENTATION else "a2a"
         return create_pipeline(
             pipeline_name,
             provider_manager=runtime.provider_manager,

src/iac_code/a2a/pipeline_snapshot.py

@@ -36,6 +36,18 @@
     PIPELINE_EVENT_CLEANUP_FAILED: "failed",
 }
 _KNOWN_CLEANUP_STATUSES = {"pending", "started", "in_progress", "completed", "failed", "skipped"}
+# Terminal statuses for steps/candidates/candidate steps inside the snapshot tree.
+# Anything else (``working``, ``pending``, ``waiting_input``, ``restarting``, missing)
+# is a dangling non-terminal node that must be finalized when the run terminates or
+# when the node gets superseded by a rollback / candidate restart.
+_TERMINAL_NODE_STATUSES = {"completed", "failed", "canceled", "superseded"}
+_SUPERSEDED_NODE_STATUS = "superseded"
+_FINALIZED_TIME_KEY_BY_STATUS = {
+    "canceled": "canceledAt",
+    "completed": "completedAt",
+    "failed": "failedAt",
+    "superseded": "supersededAt",
+}
 _PENDING_BACKUP_VISIBILITY = "pending_backup"
 _COMMITTED_BACKUP_VISIBILITY = "committed"
 _BACKUP_COMMITTED_EVENT_TYPE = "backup_committed"
@@ -501,6 +513,7 @@ def _apply(self, event: dict[str, Any]) -> None:
             self._apply_cleanup_event(event)
         elif event_type == "rollback_completed":
             self._append_rollback(event)
+            self._supersede_rolled_back_steps(event)
         elif event_type == "candidate_restart_requested":
             self._append_candidate_restart(event)
         elif event_type == "input_required":
@@ -529,6 +542,7 @@ def _apply(self, event: dict[str, Any]) -> None:
             self._snapshot["pendingTerminal"] = None
             self._snapshot["pendingInput"] = None
             self._snapshot["control"]["activeCandidateRunIds"] = []
+            self._finalize_non_terminal_nodes(terminal_status, event)
         elif (
             event_type not in {"input_required", "input_received", *_CLEANUP_STATUS_BY_EVENT_TYPE}
             and not _is_pending_backup_publication(event)
@@ -539,6 +553,109 @@ def _apply(self, event: dict[str, Any]) -> None:
         ):
             self._apply_event_status(event)
 
+    def _finalize_non_terminal_nodes(self, status: str, event: dict[str, Any]) -> None:
+        """Converge every dangling step/candidate/candidate step when the run terminates.
+
+        Only ``canceled``/``failed`` propagate. A successful run emits ``step_completed``
+        per step, so pushing ``completed`` onto a node that never reported completion
+        would fabricate a success that did not happen.
+        """
+        if status not in {"canceled", "failed"}:
+            return
+        reason = _string_or_none(event.get("eventType")) or status
+        for step in self._snapshot["steps"]:
+            _finalize_node(step, status, event, reason=reason, finalized_by="run_termination")
+            for candidate in _dict_list(step.get("candidates")):
+                _finalize_node(candidate, status, event, reason=reason, finalized_by="run_termination")
+                for candidate_step in _dict_list(candidate.get("steps")):
+                    _finalize_node(candidate_step, status, event, reason=reason, finalized_by="run_termination")
+
+    def _supersede_rolled_back_steps(self, event: dict[str, Any]) -> None:
+        """Collapse the pre-rollback attempts of the rollback target step.
+
+        ``pipeline_events`` bumps the step ``attempt`` on rollback, so the replayed step
+        lands as a new record while the interrupted attempt would otherwise stay
+        ``working`` with a ``null`` conclusion next to it.
+        """
+        coordinate = _dict_or_none(event.get("step"))
+        if coordinate is None:
+            return
+        step_id = _string_or_none(coordinate.get("id"))
+        attempt = _int_or_none(coordinate.get("attempt"))
+        if step_id is None or attempt is None:
+            return
+        for step in self._snapshot["steps"]:
+            if _string_or_none(step.get("id")) != step_id:
+                continue
+            if (_int_or_none(step.get("attempt")) or 1) >= attempt:
+                continue
+            _finalize_node(
+                step,
+                _SUPERSEDED_NODE_STATUS,
+                event,
+                reason="rollback_completed",
+                finalized_by="rollback_collapse",
+            )
+            for candidate in _dict_list(step.get("candidates")):
+                _finalize_node(
+                    candidate,
+                    _SUPERSEDED_NODE_STATUS,
+                    event,
+                    reason="rollback_completed",
+                    finalized_by="rollback_collapse",
+                )
+                for candidate_step in _dict_list(candidate.get("steps")):
+                    _finalize_node(
+                        candidate_step,
+                        _SUPERSEDED_NODE_STATUS,
+                        event,
+                        reason="rollback_completed",
+                        finalized_by="rollback_collapse",
+                    )
+
+    def _supersede_earlier_candidate_attempts(self, candidate: dict[str, Any], event: dict[str, Any]) -> None:
+        """Terminate the previous attempts of the candidate that just (re)started."""
+        candidate_id = _string_or_none(candidate.get("id"))
+        candidate_index = _int_or_none(candidate.get("index"))
+        attempt = _int_or_none(candidate.get("attempt"))
+        if candidate_id is None or candidate_index is None or attempt is None:
+            return
+        for other in self._candidates_by_run_id.values():
+            if other is candidate:
+                continue
+            if _string_or_none(other.get("id")) != candidate_id:
+                continue
+            if _int_or_none(other.get("index")) != candidate_index:
+                continue
+            if (_int_or_none(other.get("attempt")) or 1) >= attempt:
+                continue
+            _finalize_node(
+                other,
+                _SUPERSEDED_NODE_STATUS,
+                event,
+                reason="candidate_restarted",
+                finalized_by="candidate_restart",
+            )
+            for candidate_step in _dict_list(other.get("steps")):
+                _finalize_node(
+                    candidate_step,
+                    _SUPERSEDED_NODE_STATUS,
+                    event,
+                    reason="candidate_restarted",
+                    finalized_by="candidate_restart",
+                )
+
+    def _supersede_restarted_candidate(self, candidate: dict[str, Any], event: dict[str, Any]) -> None:
+        """Terminate the sub-steps of a candidate that is being restarted."""
+        for candidate_step in _dict_list(candidate.get("steps")):
+            _finalize_node(
+                candidate_step,
+                _SUPERSEDED_NODE_STATUS,
+                event,
+                reason="candidate_restart_requested",
+                finalized_by="candidate_restart",
+            )
+
     def _merge_pipeline_identity(self, event: dict[str, Any]) -> None:
         for key in ("pipelineRunId", "taskId", "contextId", "pipelineName"):
             value = event.get(key)
@@ -672,6 +789,7 @@ def _apply_candidate_lifecycle(self, candidate: dict[str, Any], event: dict[str,
             candidate["status"] = "working"
             _set_time(candidate, "startedAt", created_at)
             self._remove_active_candidate_attempts(candidate)
+            self._supersede_earlier_candidate_attempts(candidate, event)
             _append_unique(self._snapshot["control"]["activeCandidateRunIds"], run_id)
         elif event_type == "candidate_completed":
             candidate["status"] = "completed"
@@ -687,6 +805,7 @@ def _apply_candidate_lifecycle(self, candidate: dict[str, Any], event: dict[str,
             candidate["status"] = "restarting"
             _set_time(candidate, "restartingAt", created_at)
             _remove_value(self._snapshot["control"]["activeCandidateRunIds"], run_id)
+            self._supersede_restarted_candidate(candidate, event)
 
     def _remove_active_candidate_attempts(self, candidate: dict[str, Any]) -> None:
         candidate_id = _string_or_none(candidate.get("id"))
@@ -1055,6 +1174,37 @@ def _backup_blocked_pending_input(self, event: dict[str, Any]) -> dict[str, Any]
         return pending
 
 
+def _is_terminal_node(node: dict[str, Any]) -> bool:
+    return _string_or_none(node.get("status")) in _TERMINAL_NODE_STATUSES
+
+
+def _finalize_node(
+    node: dict[str, Any],
+    status: str,
+    event: dict[str, Any],
+    *,
+    reason: str,
+    finalized_by: str,
+) -> None:
+    """Move a dangling snapshot node to ``status`` and give it an explicit conclusion.
+
+    Already-terminal nodes keep their own status and conclusion, and a node that carries
+    a real conclusion never has it overwritten, so the projection stays idempotent and
+    never rewrites a result the pipeline actually reported.
+    """
+    if _is_terminal_node(node):
+        return
+    node["status"] = status
+    created_at = _string_or_none(event.get("createdAt"))
+    _set_time(node, _FINALIZED_TIME_KEY_BY_STATUS.get(status, "finalizedAt"), created_at)
+    if node.get("conclusion") is None:
+        node["conclusion"] = {
+            "status": status,
+            "terminationReason": reason,
+            "finalizedBy": finalized_by,
+        }
+
+
 def _normal_handoff(event: dict[str, Any]) -> dict[str, Any]:
     data = _sanitize_cleanup_private_fields(copy.deepcopy(_dict_or_empty(event.get("data"))))
     handoff = {

src/iac_code/a2a/task_store.py

@@ -864,9 +864,7 @@ async def has_active_work(self) -> bool:
                 any(record.active_task is not None and not record.active_task.done() for record in self._tasks.values())
                 or any(not task.done() for task in self._context_runtime_tasks.values())
                 or any(
-                    not task.done()
-                    for starts in self._context_execution_starts.values()
-                    for task in starts.values()
+                    not task.done() for starts in self._context_execution_starts.values() for task in starts.values()
                 )
                 or any(self._context_reconciliation_waiters.values())
                 or any(lock.locked() for lock in self._reconciliation_locks.values())

src/iac_code/pipeline/engine/loader.py

@@ -343,9 +343,7 @@ def _parse_surface_overrides(raw: object, step_id: str) -> dict[str, StepSurface
 
         conclusion_schema = override.get("conclusion_schema")
         if conclusion_schema is not None and not isinstance(conclusion_schema, dict):
-            raise ValueError(
-                f"Step '{step_id}': surface_overrides.{surface}.conclusion_schema must be a mapping"
-            )
+            raise ValueError(f"Step '{step_id}': surface_overrides.{surface}.conclusion_schema must be a mapping")
 
         overrides[surface] = StepSurfaceOverride(
             prompt_file=prompt,

src/iac_code/pipeline/engine/pipeline_runner.py

@@ -93,6 +93,8 @@
 
 def _is_a2a_surface(surface: str) -> bool:
     return surface == "a2a" or surface.startswith("a2a_")
+
+
 _SIDECAR_ROOT_DIRS = {"a2a", "image-cache", "pipeline", "tool-results"}
 _SIDECAR_ROOT_FILES = {
     ".backup-state.json",

src/iac_code/services/providers/aliyun.py

@@ -298,9 +298,7 @@ def refresh_oauth_if_needed(
 
             owns_client = oauth_client is None
             client = (
-                AliyunOAuthClient(get_oauth_site(credential.oauth_site_type))
-                if oauth_client is None
-                else oauth_client
+                AliyunOAuthClient(get_oauth_site(credential.oauth_site_type)) if oauth_client is None else oauth_client
             )
 
             try:

src/iac_code/services/session_backup_staging.py

@@ -149,8 +149,7 @@ def backup_session(
                         existing = self._read_existing_snapshot_state(destination, session_id)
                         if existing is not None:
                             completed_next = (
-                                base_state.status == "succeeded"
-                                and existing.parent_generation == base_state.generation
+                                base_state.status == "succeeded" and existing.parent_generation == base_state.generation
                             )
                             if not completed_next and not existing.same_lineage(committed_state):
                                 raise SessionBackupConflict(

tests/a2a/test_executor.py

@@ -3788,12 +3788,10 @@ def test_accepts_skill_rich_presentation_metadata(self) -> None:
         executor = self._make_executor()
 
         assert (
-            executor._resolve_candidate_presentation({"iac_code": {"candidatePresentation": " rich-v1 "}})
-            == "rich-v1"
+            executor._resolve_candidate_presentation({"iac_code": {"candidatePresentation": " rich-v1 "}}) == "rich-v1"
         )
         assert (
-            executor._resolve_candidate_presentation({"iac_code": {"candidate_presentation": "RICH-V1"}})
-            == "rich-v1"
+            executor._resolve_candidate_presentation({"iac_code": {"candidate_presentation": "RICH-V1"}}) == "rich-v1"
         )
 
     def test_rejects_unknown_or_missing_presentation(self) -> None:

tests/a2a/test_input_required.py

@@ -240,9 +240,7 @@ def test_ros_deployment_permission_is_localized_and_preserves_safe_plan_summary(
                         "stackName": "demo-stack",
                         "template": "templates/demo.yml",
                         "totalMonthlyCost": "¥88/月",
-                        "resources": [
-                            {"name": "ECS", "spec": "2 vCPU / 4 GiB", "monthlyCost": "¥88/月"}
-                        ],
+                        "resources": [{"name": "ECS", "spec": "2 vCPU / 4 GiB", "monthlyCost": "¥88/月"}],
                     },
                 },
             ),
@@ -388,9 +386,7 @@ async def record_before_enqueue(envelope):
 async def test_sub_pipeline_permissions_stay_working_and_resolve_independently(monkeypatch, tmp_path) -> None:
     registry = PermissionInputRegistry()
     store = A2ATaskStore()
-    await store.save(
-        Task(id="task-1", context_id="ctx-1", status=TaskStatus(state=TaskState.TASK_STATE_WORKING))
-    )
+    await store.save(Task(id="task-1", context_id="ctx-1", status=TaskStatus(state=TaskState.TASK_STATE_WORKING)))
     queue = FakeEventQueue()
     publisher = PipelineA2AEventPublisher(
         event_queue=queue,
@@ -466,9 +462,7 @@ async def test_sub_pipeline_permissions_stay_working_and_resolve_independently(m
     task = await store.get("task-1")
     assert task is not None
     task_metadata = MessageToDict(task.metadata, preserving_proto_field_name=False)
-    assert [item["inputId"] for item in task_metadata["iac_code"]["pendingPermissions"]] == [
-        requests[1]["inputId"]
-    ]
+    assert [item["inputId"] for item in task_metadata["iac_code"]["pendingPermissions"]] == [requests[1]["inputId"]]
     remaining = task_metadata["iac_code"]["pendingPermissions"][0]
     assert remaining["language"] == "zh"
     assert remaining["prompt"] == "是否允许本次操作：运行本地 Shell 命令？"