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..a65579ab 100644 --- a/src/iac_code/a2a/pipeline_executor.py +++ b/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 diff --git a/src/iac_code/a2a/pipeline_snapshot.py b/src/iac_code/a2a/pipeline_snapshot.py index fb8b9aa6..ae3780f7 100644 --- a/src/iac_code/a2a/pipeline_snapshot.py +++ b/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", 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/a2a/test_pipeline_executor.py b/tests/a2a/test_pipeline_executor.py index ccae8d3a..83d1273a 100644 --- a/tests/a2a/test_pipeline_executor.py +++ b/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"}} diff --git a/tests/a2a/test_pipeline_snapshot.py b/tests/a2a/test_pipeline_snapshot.py index a3a7e01c..8f781607 100644 --- a/tests/a2a/test_pipeline_snapshot.py +++ b/tests/a2a/test_pipeline_snapshot.py @@ -1480,3 +1480,157 @@ 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 _intent_parsing_step() -> dict: + return {"runId": "step-intent_parsing-1", "id": "intent_parsing", "index": 1, "total": 3, "attempt": 1} + + +def test_reduce_pipeline_canceled_finalizes_working_step() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + canceled = _base("evt-2", 2, "pipeline_canceled", status="canceled") + canceled["data"] = {"source": "executor", "reason": "Task canceled."} + + snapshot = reduce_pipeline_events([started, canceled]) + + assert snapshot["status"] == "canceled" + step = snapshot["steps"][0] + assert step["status"] == "canceled" + assert step["canceledAt"] == "2026-06-08T10:00:00Z" + assert step["conclusion"] == { + "pipelineTerminated": True, + "terminalStatus": "canceled", + "reason": "Task canceled.", + } + + +def test_reduce_pipeline_canceled_finalizes_candidates_and_candidate_steps() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + candidate = _base("evt-2", 2, "candidate_started", scope="candidate") + candidate["step"] = started["step"] + candidate["candidate"] = {"runId": "candidate-eval-0-1", "id": "eval", "index": 0, "attempt": 1} + candidate_step = _base("evt-3", 3, "candidate_step_started", scope="candidate_step") + candidate_step["step"] = started["step"] + candidate_step["candidate"] = candidate["candidate"] + candidate_step["candidateStep"] = { + "runId": "candidate-eval-0-1-template-1", + "id": "template", + "index": 1, + "total": 1, + "attempt": 1, + } + canceled = _base("evt-4", 4, "pipeline_canceled", status="canceled") + + snapshot = reduce_pipeline_events([started, candidate, candidate_step, canceled]) + + step = snapshot["steps"][0] + assert step["status"] == "canceled" + assert step["candidates"][0]["status"] == "canceled" + assert step["candidates"][0]["steps"][0]["status"] == "canceled" + assert step["candidates"][0]["conclusion"]["pipelineTerminated"] is True + assert step["candidates"][0]["steps"][0]["conclusion"]["pipelineTerminated"] is True + assert snapshot["control"]["activeCandidateRunIds"] == [] + + +def test_reduce_pipeline_canceled_omits_reason_when_event_has_none() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + canceled = _base("evt-2", 2, "pipeline_canceled", status="canceled") + + snapshot = reduce_pipeline_events([started, canceled]) + + assert snapshot["steps"][0]["conclusion"] == {"pipelineTerminated": True, "terminalStatus": "canceled"} + + +def test_reduce_pipeline_failed_finalizes_working_step() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + failed = _base("evt-2", 2, "pipeline_failed", status="failed") + failed["data"] = {"errorSummary": "provider unavailable"} + + snapshot = reduce_pipeline_events([started, failed]) + + step = snapshot["steps"][0] + assert step["status"] == "failed" + assert step["failedAt"] == "2026-06-08T10:00:00Z" + assert step["errorSummary"] == "provider unavailable" + assert step["conclusion"]["terminalStatus"] == "failed" + + +def test_reduce_pipeline_canceled_finalizes_waiting_input_step() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + waiting = _base("evt-2", 2, "input_required", scope="input", status="input_required") + waiting["step"] = started["step"] + canceled = _base("evt-3", 3, "pipeline_canceled", status="canceled") + + snapshot = reduce_pipeline_events([started, waiting, canceled]) + + assert snapshot["steps"][0]["status"] == "canceled" + assert snapshot["pendingInput"] is None + + +def test_reduce_pipeline_canceled_keeps_already_terminal_step_conclusion() -> None: + completed = _base("evt-1", 1, "step_completed", scope="step") + completed["step"] = _intent_parsing_step() + completed["data"] = {"conclusionField": "intent", "conclusion": {"intent": "deploy"}} + running = _base("evt-2", 2, "step_started", scope="step") + running["step"] = {"runId": "step-planning-2", "id": "planning", "index": 2, "total": 3, "attempt": 1} + canceled = _base("evt-3", 3, "pipeline_canceled", status="canceled") + + snapshot = reduce_pipeline_events([completed, running, canceled]) + + finished, interrupted = snapshot["steps"] + assert finished["status"] == "completed" + assert finished["conclusion"] == {"intent": "deploy"} + assert "canceledAt" not in finished + assert interrupted["status"] == "canceled" + assert interrupted["conclusion"]["pipelineTerminated"] is True + + +def test_reduce_pipeline_completed_does_not_fabricate_step_conclusions() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + completed = _base("evt-2", 2, "pipeline_completed", status="completed") + + snapshot = reduce_pipeline_events([started, completed]) + + assert snapshot["status"] == "completed" + assert snapshot["steps"][0]["status"] == "working" + assert "conclusion" not in snapshot["steps"][0] + + +def test_reduce_pending_backup_cancel_does_not_finalize_steps() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + canceled = _base("evt-2", 2, "pipeline_canceled", status="canceled") + canceled["visibility"] = "pending_backup" + + snapshot = reduce_pipeline_events([started, canceled]) + + assert snapshot["status"] == "working" + assert snapshot["pendingTerminal"]["eventType"] == "pipeline_canceled" + assert snapshot["steps"][0]["status"] == "working" + + +def test_reduce_pipeline_canceled_finalization_is_idempotent_across_reductions() -> None: + started = _base("evt-1", 1, "step_started", scope="step") + started["step"] = _intent_parsing_step() + canceled = _base("evt-2", 2, "pipeline_canceled", status="canceled") + canceled["data"] = {"reason": "Task canceled."} + late = _base("evt-3", 3, "pipeline_warning") + + once = reduce_pipeline_events([started, canceled]) + twice = reduce_pipeline_events([late], existing_snapshot=once) + + assert twice["steps"][0]["status"] == "canceled" + assert twice["steps"][0]["conclusion"] == once["steps"][0]["conclusion"] + + +def test_is_terminated_node_conclusion_only_matches_synthesized_conclusions() -> None: + assert pipeline_snapshot.is_terminated_node_conclusion({"pipelineTerminated": True, "terminalStatus": "canceled"}) + assert not pipeline_snapshot.is_terminated_node_conclusion({"intent": "deploy"}) + assert not pipeline_snapshot.is_terminated_node_conclusion(None) + assert "is_terminated_node_conclusion" in pipeline_snapshot.__all__ 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)连同思考、工具一起折进「已处理」,不平铺成答案。