fix(a2a): Pipeline 取消后收敛残留 working 的步骤到终态

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

@@ -31,7 +31,11 @@
     a2a_pipeline_dir_for_sidecar_dir,
     existing_a2a_pipeline_dir_for_session,
 )
-from iac_code.a2a.pipeline_snapshot import A2APipelineSnapshotStore, reduce_pipeline_events
+from iac_code.a2a.pipeline_snapshot import (
+    A2APipelineSnapshotStore,
+    is_terminated_node_conclusion,
+    reduce_pipeline_events,
+)
 from iac_code.a2a.pipeline_stream import (
     BACKUP_COMMITTED_EVENT_TYPE,
     PipelineA2AEventPublisher,
@@ -1235,11 +1239,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,
@@ -3972,7 +3972,7 @@ def _flat_pipeline_context_from_a2a_snapshot(snapshot: dict[str, Any] | None, lo
         if not field_name:
             continue
         conclusion = step.get("conclusion")
-        if conclusion is not None:
+        if conclusion is not None and not is_terminated_node_conclusion(conclusion):
             context[field_name] = conclusion
     return context
 

src/iac_code/a2a/pipeline_snapshot.py

@@ -29,6 +29,13 @@
     "pipeline_failed": "failed",
     "pipeline_canceled": "canceled",
 }
+_TERMINAL_NODE_STATUSES = {"completed", "failed", "canceled"}
+# ``completed`` is intentionally excluded: marking an unfinished step as completed would
+# fabricate a successful business conclusion for work that never ran.
+_INTERRUPTED_TERMINAL_STATUSES = {"failed", "canceled"}
+# Marks a conclusion synthesized by the reducer rather than reported by a step, so
+# consumers such as the pipeline handoff context can tell it apart from real results.
+_TERMINATED_CONCLUSION_MARKER = "pipelineTerminated"
 _CLEANUP_STATUS_BY_EVENT_TYPE = {
     PIPELINE_EVENT_CLEANUP_STARTED: "started",
     PIPELINE_EVENT_CLEANUP_PROGRESS: "in_progress",
@@ -529,6 +536,7 @@ def _apply(self, event: dict[str, Any]) -> None:
             self._snapshot["pendingTerminal"] = None
             self._snapshot["pendingInput"] = None
             self._snapshot["control"]["activeCandidateRunIds"] = []
+            self._finalize_open_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)
@@ -750,6 +758,35 @@ def _apply_candidate_step_lifecycle(self, candidate_step: dict[str, Any], event:
             _set_time(candidate_step, "failedAt", created_at)
             _merge_completion_data(candidate_step, event)
 
+    def _finalize_open_nodes(self, terminal_status: str, event: dict[str, Any]) -> None:
+        """Converge every still-open step/candidate/sub-step onto the run's terminal status.
+
+        A run interrupted mid-step never emits ``step_completed``/``step_failed`` for the
+        step it was executing, so without this the snapshot keeps that step at ``working``
+        with ``conclusion`` null forever while the run itself is already terminal.  Mirrors
+        ``web.pipeline_transcript._on_pipeline_canceled``, which finalizes the same nodes in
+        the transcript, so both projections agree on the outcome of one run.
+        """
+        if terminal_status not in _INTERRUPTED_TERMINAL_STATUSES:
+            return
+
+        created_at = _string_or_none(event.get("createdAt"))
+        time_key = "canceledAt" if terminal_status == "canceled" else "failedAt"
+        reason = _string_or_none(_dict_or_empty(event.get("data")).get("reason"))
+        for nodes in (
+            self._steps_by_run_id,
+            self._candidates_by_run_id,
+            self._candidate_steps_by_run_id,
+        ):
+            for node in nodes.values():
+                if _string_or_none(node.get("status")) in _TERMINAL_NODE_STATUSES:
+                    continue
+                node["status"] = terminal_status
+                _set_time(node, time_key, created_at)
+                _merge_completion_data(node, event)
+                if node.get("conclusion") is None:
+                    node["conclusion"] = _terminated_node_conclusion(terminal_status, reason)
+
     def _apply_text_delta(self, event: dict[str, Any]) -> None:
         text = _dict_or_empty(event.get("data")).get("text")
         if not isinstance(text, str):
@@ -1741,6 +1778,18 @@ def _set_time(target: dict[str, Any], key: str, value: str | None) -> None:
         target[key] = value
 
 
+def _terminated_node_conclusion(terminal_status: str, reason: str | None) -> dict[str, Any]:
+    conclusion: dict[str, Any] = {_TERMINATED_CONCLUSION_MARKER: True, "terminalStatus": terminal_status}
+    if reason is not None:
+        conclusion["reason"] = reason
+    return conclusion
+
+
+def is_terminated_node_conclusion(conclusion: Any) -> bool:
+    """Report whether a snapshot node conclusion was synthesized on pipeline termination."""
+    return isinstance(conclusion, dict) and conclusion.get(_TERMINATED_CONCLUSION_MARKER) is True
+
+
 def _merge_completion_data(target: dict[str, Any], event: dict[str, Any]) -> None:
     data = _dict_or_empty(event.get("data"))
     for key in (
@@ -1775,6 +1824,7 @@ def _utc_now() -> str:
 __all__ = [
     "A2APipelineSnapshotStore",
     "SNAPSHOT_SCHEMA_VERSION",
+    "is_terminated_node_conclusion",
     "reduce_pipeline_events",
     "sanitize_pipeline_cleanup_private_fields",
     "snapshot_needs_backup_commit_repair",

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 命令？"

tests/a2a/test_pipeline_executor.py

@@ -10759,3 +10759,28 @@ async def resume_ask_user_question(self, answer, **kwargs):
 
     assert events
     assert received["supplemental_input"] == pipeline_input
+
+
+def test_flat_pipeline_context_from_snapshot_skips_terminated_step_conclusions() -> None:
+    from iac_code.a2a.pipeline_executor import _flat_pipeline_context_from_a2a_snapshot
+
+    loaded_pipeline = SimpleNamespace(
+        steps=[
+            SimpleNamespace(step_id="intent_parsing", conclusion_field="intent"),
+            SimpleNamespace(step_id="planning", conclusion_field="plan"),
+        ]
+    )
+    snapshot = {
+        "steps": [
+            {"id": "intent_parsing", "status": "completed", "conclusion": {"intent": "deploy"}},
+            {
+                "id": "planning",
+                "status": "canceled",
+                "conclusion": {"pipelineTerminated": True, "terminalStatus": "canceled"},
+            },
+        ]
+    }
+
+    context = _flat_pipeline_context_from_a2a_snapshot(snapshot, loaded_pipeline)
+
+    assert context == {"intent": {"intent": "deploy"}}