From 73595caf57abac1daf0f691d4ad06f0064c8b433 Mon Sep 17 00:00:00 2001 From: "ruanzhengxin.rzx" Date: Tue, 25 Aug 2026 15:18:39 +0800 Subject: [PATCH 1/2] chore: normalize formatting with the resolved ruff version MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pyproject 里 ruff 未固定版本（ruff>=0.4.0），uv 解析到 0.15.10，其 formatter 与当前提交树最后一次格式化所用版本不一致，导致 `make format` 每次都会重写这 16 个与本次改动无关的文件；由于 format 钩子是全树执行（pass_filenames: false），任何提交都会被它拦下。 这里只落地 `make format` 的机械输出，不含任何逻辑改动，以便后续修复的 diff 保持可评审。 --- src/iac_code/a2a/executor.py | 4 +-- src/iac_code/a2a/input_required.py | 26 ++++++----------- src/iac_code/a2a/pipeline_executor.py | 6 +--- src/iac_code/a2a/task_store.py | 4 +-- src/iac_code/pipeline/engine/loader.py | 4 +-- .../pipeline/engine/pipeline_runner.py | 2 ++ src/iac_code/services/providers/aliyun.py | 4 +-- .../services/session_backup_staging.py | 3 +- tests/a2a/test_executor.py | 6 ++-- tests/a2a/test_input_required.py | 12 ++------ tests/pipeline/engine/test_step_executor.py | 28 ++++++++++++------- .../selling/test_terminal_ui_contract.py | 4 +-- tests/providers/test_manager.py | 4 +-- tests/skill_bridge/test_iac_code_bridge.py | 10 ++----- tests/skill_bridge/test_runtime_release.py | 4 +-- tests/web/test_frontend_static.py | 2 +- 16 files changed, 46 insertions(+), 77 deletions(-) diff --git a/src/iac_code/a2a/executor.py b/src/iac_code/a2a/executor.py index 226a2d44..8c642210 100644 --- a/src/iac_code/a2a/executor.py +++ b/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(): diff --git a/src/iac_code/a2a/input_required.py b/src/iac_code/a2a/input_required.py index 4418de08..c0ba54a3 100644 --- a/src/iac_code/a2a/input_required.py +++ b/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 "" diff --git a/src/iac_code/a2a/pipeline_executor.py b/src/iac_code/a2a/pipeline_executor.py index cf073fec..127cb079 100644 --- a/src/iac_code/a2a/pipeline_executor.py +++ b/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, diff --git a/src/iac_code/a2a/task_store.py b/src/iac_code/a2a/task_store.py index 0f1ae69d..a0906315 100644 --- a/src/iac_code/a2a/task_store.py +++ b/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()) diff --git a/src/iac_code/pipeline/engine/loader.py b/src/iac_code/pipeline/engine/loader.py index 88f5e298..1d2ee697 100644 --- a/src/iac_code/pipeline/engine/loader.py +++ b/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, diff --git a/src/iac_code/pipeline/engine/pipeline_runner.py b/src/iac_code/pipeline/engine/pipeline_runner.py index 59b22a03..a7b4c075 100644 --- a/src/iac_code/pipeline/engine/pipeline_runner.py +++ b/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", diff --git a/src/iac_code/services/providers/aliyun.py b/src/iac_code/services/providers/aliyun.py index a66f6d59..e0119007 100644 --- a/src/iac_code/services/providers/aliyun.py +++ b/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: diff --git a/src/iac_code/services/session_backup_staging.py b/src/iac_code/services/session_backup_staging.py index 6d061c47..e5d933a3 100644 --- a/src/iac_code/services/session_backup_staging.py +++ b/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( diff --git a/tests/a2a/test_executor.py b/tests/a2a/test_executor.py index 4c48e2b7..4e0a31e1 100644 --- a/tests/a2a/test_executor.py +++ b/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: diff --git a/tests/a2a/test_input_required.py b/tests/a2a/test_input_required.py index b7615e0e..0f310aa7 100644 --- a/tests/a2a/test_input_required.py +++ b/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 命令？" diff --git a/tests/pipeline/engine/test_step_executor.py b/tests/pipeline/engine/test_step_executor.py index 2d596efe..c85dc073 100644 --- a/tests/pipeline/engine/test_step_executor.py +++ b/tests/pipeline/engine/test_step_executor.py @@ -2776,11 +2776,15 @@ def test_rich_candidate_resume_uses_compact_schema_and_preserves_first_conclusio surface="a2a_rich", ) - tool_schema = executor._build_step_tools( - step, - context, - compact_candidate_selection=True, - ).get("complete_step").input_schema + tool_schema = ( + executor._build_step_tools( + step, + context, + compact_candidate_selection=True, + ) + .get("complete_step") + .input_schema + ) conclusion_schema = tool_schema["properties"]["conclusion"] assert conclusion_schema["required"] == [ "selected_candidate_name", @@ -2863,11 +2867,15 @@ def test_stale_candidate_conclusion_cannot_enable_compact_resume_schema(self, tm context, resume_candidate_selection=True, ) - tool_schema = executor._build_step_tools( - step, - context, - compact_candidate_selection=preserved is not None, - ).get("complete_step").input_schema + tool_schema = ( + executor._build_step_tools( + step, + context, + compact_candidate_selection=preserved is not None, + ) + .get("complete_step") + .input_schema + ) assert preserved is None conclusion_schema = tool_schema["properties"]["conclusion"] diff --git a/tests/pipeline/selling/test_terminal_ui_contract.py b/tests/pipeline/selling/test_terminal_ui_contract.py index c4e2cd6a..ad9382b7 100644 --- a/tests/pipeline/selling/test_terminal_ui_contract.py +++ b/tests/pipeline/selling/test_terminal_ui_contract.py @@ -133,9 +133,7 @@ def test_confirm_prompt_tells_model_to_preserve_parameter_overrides(): def test_confirm_prompts_share_selection_contract_structure(): repl_prompt = (_selling_pipeline_dir() / "prompts" / "confirm_and_select.md").read_text(encoding="utf-8") a2a_prompt = (_selling_pipeline_dir() / "prompts" / "confirm_and_select.a2a.md").read_text(encoding="utf-8") - rich_prompt = (_selling_pipeline_dir() / "prompts" / "confirm_and_select.a2a.rich.md").read_text( - encoding="utf-8" - ) + rich_prompt = (_selling_pipeline_dir() / "prompts" / "confirm_and_select.a2a.rich.md").read_text(encoding="utf-8") shared_fragments = [ "## 首次执行", diff --git a/tests/providers/test_manager.py b/tests/providers/test_manager.py index 8ff0fc1a..54057e5f 100644 --- a/tests/providers/test_manager.py +++ b/tests/providers/test_manager.py @@ -1676,9 +1676,7 @@ async def test_complete_records_chat_span_event_and_total_metric(self): async def test_complete_attributes_bailian_openai_endpoint_to_dashscope_on_all_signals(self): mock_provider = AsyncMock() - mock_provider._base_url = ( - "https://llm-testworkspace000000.cn-beijing.maas.aliyuncs.com/compatible-mode/v1" - ) + mock_provider._base_url = "https://llm-testworkspace000000.cn-beijing.maas.aliyuncs.com/compatible-mode/v1" mock_provider.complete = AsyncMock( return_value=NonStreamingResponse( message_id="complete-response", diff --git a/tests/skill_bridge/test_iac_code_bridge.py b/tests/skill_bridge/test_iac_code_bridge.py index b7e9280f..1eafe79e 100644 --- a/tests/skill_bridge/test_iac_code_bridge.py +++ b/tests/skill_bridge/test_iac_code_bridge.py @@ -542,12 +542,10 @@ def finish_cleanup(_args): assert len(captured_payloads) == 2 assert all( - payload["params"]["message"]["metadata"]["iac_code"]["cleanupOnly"] is True - for payload in captured_payloads + payload["params"]["message"]["metadata"]["iac_code"]["cleanupOnly"] is True for payload in captured_payloads ) assert all( - payload["params"]["message"]["metadata"]["iac_code"]["channel"] == "skill/host" - for payload in captured_payloads + payload["params"]["message"]["metadata"]["iac_code"]["channel"] == "skill/host" for payload in captured_payloads ) assert captured_payloads[0]["params"]["message"]["contextId"] == "ctx-pipeline-1" assert result["state"] == "completed" @@ -1138,9 +1136,7 @@ def test_candidate_presentation_survives_bounded_bridge_projection() -> None: "summary": "单 ECS 低成本方案。", "architectureDiagram": "flowchart LR\nU[用户] --> E[ECS]", "totalMonthlyCost": "¥88/月", - "costItems": [ - {"name": "ECS", "spec": "2核4G", "monthlyCost": "¥88/月"} - ], + "costItems": [{"name": "ECS", "spec": "2核4G", "monthlyCost": "¥88/月"}], } ], "required": True, diff --git a/tests/skill_bridge/test_runtime_release.py b/tests/skill_bridge/test_runtime_release.py index 140ebb59..d6de0578 100644 --- a/tests/skill_bridge/test_runtime_release.py +++ b/tests/skill_bridge/test_runtime_release.py @@ -104,9 +104,7 @@ def test_runtime_archive_and_version_marker_are_rooted_consistently(tmp_path: Pa assert "artifactRevision" not in marker -def test_runtime_a2a_smoke_checks_health_and_agent_card( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_runtime_a2a_smoke_checks_health_and_agent_card(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: module = _load_module("skill_runtime_smoke", BUILD_SCRIPT) server = tmp_path / "server.py" server.write_text( diff --git a/tests/web/test_frontend_static.py b/tests/web/test_frontend_static.py index c3c6f30f..fe9e6f69 100644 --- a/tests/web/test_frontend_static.py +++ b/tests/web/test_frontend_static.py @@ -2547,7 +2547,7 @@ def test_completed_turn_collapses_process_into_summary() -> None: # 「已处理」组的展开态必须跨重建保留:openKey 让 toggle 记录器登记用户操作、 # applyDetailsOpenOverrides 在重建后恢复;键取 turnId,缺 turnId 时回退首条消息 id。 assert 'const turnKey = turnId || text(agentMessages[0]?.messageId || agentMessages[0]?.id || "");' in app_source - assert 'details.dataset.openKey = `turnproc:${turnKey}`;' in app_source + assert "details.dataset.openKey = `turnproc:${turnKey}`;" in app_source # 只有最后一次工具调用之后的文本才是「最终回答」;此前每个步骤的文本旁白 # (夹在工具调用之间的 text delta)连同思考、工具一起折进「已处理」,不平铺成答案。 From cba5a07d4ba262428970a9cb8a3ef4d27f54619b Mon Sep 17 00:00:00 2001 From: "ruanzhengxin.rzx" Date: Tue, 25 Aug 2026 15:19:08 +0800 Subject: [PATCH 2/2] fix(a2a): finalize dangling pipeline snapshot nodes on cancel/rollback/restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 取消、回滚、候选重跑之后，a2a-snapshot.json 中被打断的 step/candidate/ candidate step 仍停留在 working 且 conclusion 为 null，前端因此永远渲染 "进行中"，且同一逻辑步骤会出现新旧两条非终态记录。 根因在 _PipelineSnapshotReducer：pipeline_canceled / pipeline_failed 只 改写快照顶层 status 并清空 activeCandidateRunIds，从不下钻步骤树； rollback_completed 只追加历史；candidate_restart_requested 只把候选置为 非终态的 restarting，都不会终结被取代的节点。 - 引入 _TERMINAL_NODE_STATUSES / superseded 词表与 _finalize_node，仅在 节点尚未终态且 conclusion 为空时补写状态、时间与显式 conclusion； - 运行终止时下钻整棵步骤树收敛悬挂节点，仅传播 canceled/failed，避免为 从未上报 step_completed 的节点伪造成功； - rollback_completed 将同一 step id 的旧 attempt 折叠为 superseded； - candidate_started 终结同一候选更早的 attempt， candidate_restart_requested 终结被重启候选的子步骤。 语义与 web/pipeline_transcript.py 的 _finalize_active_markers 对齐；归约 保持幂等，历史快照重放即自愈，schemaVersion 不变、无需数据迁移。 新增 7 条回归用例，分别覆盖三个证据 Session（afb47b236ee54db5a22d295609b69de6 取消、e0c70d3197184b49826469b509a6fd21 回滚、7eaff0d9bcff4de7a1ba510ed030f017 候选重启）、成功运行不被伪造、已终态节点不被覆盖与重放幂等。 --- src/iac_code/a2a/pipeline_snapshot.py | 150 ++++++++++++++++++++++++++ tests/a2a/test_pipeline_snapshot.py | 133 ++++++++++++++++++++++- 2 files changed, 282 insertions(+), 1 deletion(-) diff --git a/src/iac_code/a2a/pipeline_snapshot.py b/src/iac_code/a2a/pipeline_snapshot.py index fb8b9aa6..513cdca1 100644 --- a/src/iac_code/a2a/pipeline_snapshot.py +++ b/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 = { diff --git a/tests/a2a/test_pipeline_snapshot.py b/tests/a2a/test_pipeline_snapshot.py index a3a7e01c..df219c29 100644 --- a/tests/a2a/test_pipeline_snapshot.py +++ b/tests/a2a/test_pipeline_snapshot.py @@ -1338,7 +1338,9 @@ def test_reduce_candidate_restart_removes_old_run_from_active_when_next_attempt_ candidates = snapshot["steps"][0]["candidates"] assert [candidate["runId"] for candidate in candidates] == ["candidate-eval-0-1", "candidate-eval-0-2"] - assert candidates[0]["status"] == "restarting" + # Once attempt 2 starts, attempt 1 is superseded instead of dangling on ``restarting``. + assert candidates[0]["status"] == "superseded" + assert candidates[0].get("conclusion") is not None assert candidates[1]["status"] == "working" assert snapshot["control"]["activeCandidateRunIds"] == ["candidate-eval-0-2"] @@ -1480,3 +1482,132 @@ def test_store_returns_none_for_invalid_utf8_snapshot(tmp_path) -> None: def test_snapshot_schema_version_is_exported() -> None: assert SNAPSHOT_SCHEMA_VERSION == "1.1" assert "SNAPSHOT_SCHEMA_VERSION" in pipeline_snapshot.__all__ + + +def test_reduce_pipeline_canceled_finalizes_dangling_nodes() -> None: + """Cancel evidence (session afb47b236ee54db5a22d295609b69de6): an in-flight step + used to keep ``working``/``conclusion: null`` forever after the run was canceled.""" + step = _base("evt-step", 1, "step_started", scope="step") + step["step"] = {"runId": "step-architecture_planning-1", "id": "architecture_planning", "attempt": 1} + candidate_started = _base("evt-cand", 2, "candidate_started", scope="candidate") + candidate_started["step"] = step["step"] + candidate_started["candidate"] = {"runId": "candidate-plan-0-1", "id": "plan", "index": 0, "attempt": 1} + candidate_step = _base("evt-cand-step", 3, "candidate_step_started", scope="candidate_step") + candidate_step["step"] = step["step"] + candidate_step["candidate"] = candidate_started["candidate"] + candidate_step["candidateStep"] = {"runId": "candidate-plan-0-1-template-1", "id": "template", "attempt": 1} + canceled = _base("evt-cancel", 4, "pipeline_canceled", status="canceled") + + snapshot = reduce_pipeline_events([step, candidate_started, candidate_step, canceled]) + + assert snapshot["status"] == "canceled" + tree_step = snapshot["steps"][0] + assert tree_step["status"] == "canceled" + assert tree_step.get("conclusion") is not None + assert tree_step["conclusion"]["terminationReason"] == "pipeline_canceled" + tree_candidate = tree_step["candidates"][0] + assert tree_candidate["status"] == "canceled" + assert tree_candidate.get("conclusion") is not None + assert tree_candidate["steps"][0]["status"] == "canceled" + assert tree_candidate["steps"][0].get("conclusion") is not None + assert snapshot["control"]["activeCandidateRunIds"] == [] + + +def test_reduce_pipeline_failed_finalizes_dangling_nodes() -> None: + step = _base("evt-step", 1, "step_started", scope="step") + step["step"] = {"runId": "step-a-1", "id": "a", "attempt": 1} + failed = _base("evt-failed", 2, "pipeline_failed", status="failed") + + snapshot = reduce_pipeline_events([step, failed]) + + assert snapshot["status"] == "failed" + assert snapshot["steps"][0]["status"] == "failed" + assert snapshot["steps"][0]["conclusion"]["status"] == "failed" + + +def test_reduce_pipeline_completed_does_not_fabricate_step_success() -> None: + step = _base("evt-step", 1, "step_started", scope="step") + step["step"] = {"runId": "step-a-1", "id": "a", "attempt": 1} + completed = _base("evt-done", 2, "pipeline_completed", status="completed") + + snapshot = reduce_pipeline_events([step, completed]) + + assert snapshot["status"] == "completed" + assert snapshot["steps"][0]["status"] == "working" + assert snapshot["steps"][0].get("conclusion") is None + + +def test_reduce_rollback_completed_supersedes_previous_step_attempt() -> None: + """Rollback evidence (session e0c70d3197184b49826469b509a6fd21): the interrupted + attempt stayed ``working`` next to the replayed attempt of the same step.""" + first = _base("evt-step-1", 1, "step_started", scope="step") + first["step"] = {"runId": "step-a-1", "id": "a", "attempt": 1} + candidate_started = _base("evt-cand", 2, "candidate_started", scope="candidate") + candidate_started["step"] = first["step"] + candidate_started["candidate"] = {"runId": "candidate-a-0-1", "id": "a-cand", "index": 0, "attempt": 1} + rollback = _base("evt-rollback", 3, "rollback_completed") + rollback["step"] = {"runId": "step-a-2", "id": "a", "attempt": 2} + second = _base("evt-step-2", 4, "step_started", scope="step") + second["step"] = rollback["step"] + + snapshot = reduce_pipeline_events([first, candidate_started, rollback, second]) + + steps = snapshot["steps"] + assert [step["runId"] for step in steps] == ["step-a-1", "step-a-2"] + assert steps[0]["status"] == "superseded" + assert steps[0]["conclusion"]["terminationReason"] == "rollback_completed" + assert steps[0]["candidates"][0]["status"] == "superseded" + assert steps[1]["status"] == "working" + assert steps[1].get("conclusion") is None + + +def test_reduce_candidate_restart_finalizes_candidate_sub_steps() -> None: + """Candidate restart evidence (session 7eaff0d9bcff4de7a1ba510ed030f017): the + superseded candidate kept ``working``/``pending`` sub-steps.""" + step = _base("evt-step", 1, "step_started", scope="step") + step["step"] = {"runId": "step-evaluate_candidates-1", "id": "evaluate_candidates", "attempt": 1} + candidate_started = _base("evt-cand", 2, "candidate_started", scope="candidate") + candidate_started["step"] = step["step"] + candidate_started["candidate"] = {"runId": "candidate-eval-0-1", "id": "eval", "index": 0, "attempt": 1} + candidate_step = _base("evt-cand-step", 3, "candidate_step_started", scope="candidate_step") + candidate_step["step"] = step["step"] + candidate_step["candidate"] = candidate_started["candidate"] + candidate_step["candidateStep"] = {"runId": "candidate-eval-0-1-template-1", "id": "template", "attempt": 1} + restart = _base("evt-restart", 4, "candidate_restart_requested", scope="candidate") + restart["step"] = step["step"] + restart["candidate"] = candidate_started["candidate"] + + snapshot = reduce_pipeline_events([step, candidate_started, candidate_step, restart]) + + candidate = snapshot["steps"][0]["candidates"][0] + assert candidate["status"] == "restarting" + sub_step = candidate["steps"][0] + assert sub_step["status"] == "superseded" + assert sub_step["conclusion"]["terminationReason"] == "candidate_restart_requested" + assert snapshot["control"]["activeCandidateRunIds"] == [] + + +def test_finalize_keeps_existing_terminal_status_and_conclusion() -> None: + step = _base("evt-step", 1, "step_started", scope="step") + step["step"] = {"runId": "step-a-1", "id": "a", "attempt": 1} + step_failed = _base("evt-step-failed", 2, "step_failed", scope="step", status="failed") + step_failed["step"] = step["step"] + step_failed["data"] = {"error": "boom"} + canceled = _base("evt-cancel", 3, "pipeline_canceled", status="canceled") + + snapshot = reduce_pipeline_events([step, step_failed, canceled]) + + assert snapshot["status"] == "canceled" + assert snapshot["steps"][0]["status"] == "failed" + + +def test_finalize_non_terminal_nodes_is_idempotent_on_replay() -> None: + step = _base("evt-step", 1, "step_started", scope="step") + step["step"] = {"runId": "step-a-1", "id": "a", "attempt": 1} + canceled = _base("evt-cancel", 2, "pipeline_canceled", status="canceled") + + once = reduce_pipeline_events([step, canceled]) + twice = reduce_pipeline_events([step, canceled], existing_snapshot=once) + + assert twice["steps"] == once["steps"] + assert twice["status"] == "canceled"