From e8d3209137886e9ef337359c6bfa9baebcf61b34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=A1=82=E9=A9=AC?= Date: Wed, 8 Jul 2026 22:48:33 +0800 Subject: [PATCH 1/3] Preserve ROS deploy progress after permission prompt --- .../i18n/locales/de/LC_MESSAGES/messages.po | 24 +++--- .../i18n/locales/es/LC_MESSAGES/messages.po | 24 +++--- .../i18n/locales/fr/LC_MESSAGES/messages.po | 24 +++--- .../i18n/locales/ja/LC_MESSAGES/messages.po | 24 +++--- .../i18n/locales/pt/LC_MESSAGES/messages.po | 24 +++--- .../i18n/locales/zh/LC_MESSAGES/messages.po | 24 +++--- .../pipeline/engine/display_replay.py | 42 +++++++++++ src/iac_code/ui/pipeline_display_replay.py | 34 +++++++++ src/iac_code/ui/renderer.py | 27 ++++++- src/iac_code/ui/repl.py | 17 ++++- tests/pipeline/engine/test_display_replay.py | 33 +++++++++ .../test_pipeline_display_replay_renderer.py | 33 +++++++++ tests/ui/test_renderer_events.py | 73 +++++++++++++++++++ tests/ui/test_repl_pipeline_display_replay.py | 67 ++++++++++++++++- 14 files changed, 394 insertions(+), 76 deletions(-) diff --git a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po index d094c834..96513837 100644 --- a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po @@ -5300,6 +5300,18 @@ msgstr " ✓ {name}: abgeschlossen" msgid "Rolled back" msgstr "Zurückgesetzt" +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Resource" +msgstr "Ressource" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Type" +msgstr "Typ" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Status" +msgstr "Status" + #: src/iac_code/ui/pipeline_display_replay.py msgid "Candidate" msgstr "Kandidat" @@ -5385,18 +5397,6 @@ msgstr "Nachgedacht für {seconds:.1f}s" msgid "(ctrl+o to expand)" msgstr "(ctrl+o zum Aufklappen)" -#: src/iac_code/ui/renderer.py -msgid "Resource" -msgstr "Ressource" - -#: src/iac_code/ui/renderer.py -msgid "Type" -msgstr "Typ" - -#: src/iac_code/ui/renderer.py -msgid "Status" -msgstr "Status" - #: src/iac_code/ui/renderer.py msgid "Account ID" msgstr "Konto-ID" diff --git a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po index be858d42..92a3d2ee 100644 --- a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po @@ -5286,6 +5286,18 @@ msgstr " ✓ {name}: completado" msgid "Rolled back" msgstr "Revertido" +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Resource" +msgstr "Recurso" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Type" +msgstr "Tipo" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Status" +msgstr "Estado" + #: src/iac_code/ui/pipeline_display_replay.py msgid "Candidate" msgstr "Candidato" @@ -5371,18 +5383,6 @@ msgstr "Razonamiento durante {seconds:.1f} s" msgid "(ctrl+o to expand)" msgstr "(ctrl+o para expandir)" -#: src/iac_code/ui/renderer.py -msgid "Resource" -msgstr "Recurso" - -#: src/iac_code/ui/renderer.py -msgid "Type" -msgstr "Tipo" - -#: src/iac_code/ui/renderer.py -msgid "Status" -msgstr "Estado" - #: src/iac_code/ui/renderer.py msgid "Account ID" msgstr "ID de cuenta" diff --git a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po index 6f2a4cab..894b6cd9 100644 --- a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po @@ -5305,6 +5305,18 @@ msgstr " ✓ {name} : terminé" msgid "Rolled back" msgstr "Revenu en arrière" +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Resource" +msgstr "Ressource" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Type" +msgstr "Type" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Status" +msgstr "État" + #: src/iac_code/ui/pipeline_display_replay.py msgid "Candidate" msgstr "Candidat" @@ -5390,18 +5402,6 @@ msgstr "Réflexion pendant {seconds:.1f}s" msgid "(ctrl+o to expand)" msgstr "(ctrl+o pour développer)" -#: src/iac_code/ui/renderer.py -msgid "Resource" -msgstr "Ressource" - -#: src/iac_code/ui/renderer.py -msgid "Type" -msgstr "Type" - -#: src/iac_code/ui/renderer.py -msgid "Status" -msgstr "État" - #: src/iac_code/ui/renderer.py msgid "Account ID" msgstr "Identifiant de compte" diff --git a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po index f5a330b4..b3b16fbd 100644 --- a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po @@ -5035,6 +5035,18 @@ msgstr " ✓ {name}: 完了" msgid "Rolled back" msgstr "ロールバック済み" +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Resource" +msgstr "リソース" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Type" +msgstr "種類" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Status" +msgstr "状態" + #: src/iac_code/ui/pipeline_display_replay.py msgid "Candidate" msgstr "候補" @@ -5120,18 +5132,6 @@ msgstr "{seconds:.1f} 秒考えました" msgid "(ctrl+o to expand)" msgstr "（ctrl+o で展開）" -#: src/iac_code/ui/renderer.py -msgid "Resource" -msgstr "リソース" - -#: src/iac_code/ui/renderer.py -msgid "Type" -msgstr "種類" - -#: src/iac_code/ui/renderer.py -msgid "Status" -msgstr "状態" - #: src/iac_code/ui/renderer.py msgid "Account ID" msgstr "アカウント ID" diff --git a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po index 0eceb937..e67ab9bd 100644 --- a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po @@ -5241,6 +5241,18 @@ msgstr " ✓ {name}: concluído" msgid "Rolled back" msgstr "Revertido" +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Resource" +msgstr "Recurso" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Type" +msgstr "Tipo" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Status" +msgstr "Status" + #: src/iac_code/ui/pipeline_display_replay.py msgid "Candidate" msgstr "Candidato" @@ -5326,18 +5338,6 @@ msgstr "Raciocínio por {seconds:.1f}s" msgid "(ctrl+o to expand)" msgstr "(ctrl+o para expandir)" -#: src/iac_code/ui/renderer.py -msgid "Resource" -msgstr "Recurso" - -#: src/iac_code/ui/renderer.py -msgid "Type" -msgstr "Tipo" - -#: src/iac_code/ui/renderer.py -msgid "Status" -msgstr "Status" - #: src/iac_code/ui/renderer.py msgid "Account ID" msgstr "ID da conta" diff --git a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po index e30180f8..7dc81367 100644 --- a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po @@ -4966,6 +4966,18 @@ msgstr " ✓ {name}: 已完成" msgid "Rolled back" msgstr "已回滚" +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Resource" +msgstr "资源" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Type" +msgstr "类型" + +#: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py +msgid "Status" +msgstr "状态" + #: src/iac_code/ui/pipeline_display_replay.py msgid "Candidate" msgstr "候选方案" @@ -5051,18 +5063,6 @@ msgstr "思考完成（耗时 {seconds:.1f}s）" msgid "(ctrl+o to expand)" msgstr "(ctrl+o 展开)" -#: src/iac_code/ui/renderer.py -msgid "Resource" -msgstr "资源" - -#: src/iac_code/ui/renderer.py -msgid "Type" -msgstr "类型" - -#: src/iac_code/ui/renderer.py -msgid "Status" -msgstr "状态" - #: src/iac_code/ui/renderer.py msgid "Account ID" msgstr "账号 ID" diff --git a/src/iac_code/pipeline/engine/display_replay.py b/src/iac_code/pipeline/engine/display_replay.py index 432d8999..9c4a3677 100644 --- a/src/iac_code/pipeline/engine/display_replay.py +++ b/src/iac_code/pipeline/engine/display_replay.py @@ -22,6 +22,16 @@ class DisplayToolUse: tool_use_id: str = "" +@dataclass +class DisplayStackProgress: + stack_id: str = "" + stack_name: str = "" + status: str = "" + progress_percentage: float = 0.0 + resources: list[dict[str, Any]] = field(default_factory=list) + elapsed_seconds: int = 0 + + @dataclass class DisplayCandidate: name: str @@ -83,6 +93,7 @@ class DisplayAttempt: rollback_reason: str = "" error: str = "" tools: list[DisplayToolUse] = field(default_factory=list) + stack_progress: DisplayStackProgress | None = None sub_pipelines: dict[str, DisplaySubPipeline] = field(default_factory=dict) candidate_selection: DisplayCandidateSelection = field(default_factory=DisplayCandidateSelection) @@ -152,6 +163,21 @@ def record_tool_use(self, event: Any, *, step_id: str | None = None, sub_pipelin payload["sub_pipeline_id"] = sub_pipeline_id self.record("tool_used", step_id=step_id, payload=payload) + def record_stack_progress(self, event: Any, *, step_id: str | None = None) -> None: + resources = getattr(event, "resources", []) + self.record( + "stack_progress", + step_id=step_id, + payload={ + "stack_id": getattr(event, "stack_id", ""), + "stack_name": getattr(event, "stack_name", ""), + "status": getattr(event, "status", ""), + "progress_percentage": getattr(event, "progress_percentage", 0.0), + "elapsed_seconds": getattr(event, "elapsed_seconds", 0), + "resources": resources if isinstance(resources, list) else [], + }, + ) + def record_candidate_diagram(self, event: Any, *, step_id: str | None = None) -> None: self.record( "candidate_diagram", @@ -314,6 +340,22 @@ def reduce( attempt.tools.append(DisplayToolUse(name=name, tool_use_id=str(payload.get("tool_use_id") or ""))) continue + if event_type == "stack_progress" and attempt is not None: + resources = payload.get("resources") + attempt.stack_progress = DisplayStackProgress( + stack_id=str(payload.get("stack_id") or ""), + stack_name=str(payload.get("stack_name") or ""), + status=str(payload.get("status") or ""), + progress_percentage=self._optional_float(payload.get("progress_percentage")) or 0.0, + elapsed_seconds=self._optional_int(payload.get("elapsed_seconds")) or 0, + resources=[ + dict(resource) + for resource in (resources if isinstance(resources, list) else []) + if isinstance(resource, dict) + ], + ) + continue + if event_type == "user_input_required" and attempt is not None: attempt.status = "waiting_input" self._mark_candidate_selection_waiting(attempt, payload) diff --git a/src/iac_code/ui/pipeline_display_replay.py b/src/iac_code/ui/pipeline_display_replay.py index de175f4d..8b590641 100644 --- a/src/iac_code/ui/pipeline_display_replay.py +++ b/src/iac_code/ui/pipeline_display_replay.py @@ -23,6 +23,7 @@ DisplaySubPipeline, DisplaySubStepAttempt, ) +from iac_code.tools.cloud.types import translate_status from iac_code.ui.diagram_rendering import style_attachment_lines from iac_code.ui.pipeline_styles import pipeline_step_header, pipeline_title @@ -75,6 +76,9 @@ def _render_attempt(self, attempt: DisplayAttempt) -> None: for tool in attempt.tools: self.console.print(f" ● {display_tool_name(tool.name)}") + if attempt.stack_progress is not None: + self._render_stack_progress(attempt.stack_progress) + if attempt.step_type == "parallel_sub_pipeline" or attempt.sub_pipelines: self._render_sub_pipelines(attempt) @@ -193,6 +197,36 @@ def _render_sub_step_attempt(self, sub_step: DisplaySubStepAttempt) -> None: self.console.print(message) self._replay_transcript(sub_step.transcript_id) + def _render_stack_progress(self, progress: Any) -> None: + stack_status = translate_status(str(getattr(progress, "status", "") or "")) + stack_name = str(getattr(progress, "stack_name", "") or "") + stack_id = str(getattr(progress, "stack_id", "") or "") + progress_percentage = float(getattr(progress, "progress_percentage", 0.0) or 0.0) + label = stack_name or stack_id + if stack_name and stack_id: + label = f"{stack_name}({stack_id})" + if label: + self.console.print(f" Stack: {label} [{stack_status}] {progress_percentage:.0f}%") + + resources = getattr(progress, "resources", []) + if not isinstance(resources, list) or not resources: + return + table = Table(show_header=True, border_style="dim") + table.add_column(_("Resource")) + table.add_column(_("Type")) + table.add_column(_("Status")) + for resource in resources: + if not isinstance(resource, dict): + continue + status_icon = str(resource.get("status_icon") or "") + status = str(resource.get("status") or "") + table.add_row( + str(resource.get("name") or ""), + str(resource.get("resource_type") or ""), + f"{status_icon} {translate_status(status)}".strip(), + ) + self.console.print(table) + def _render_candidate_selection(self, selection: DisplayCandidateSelection) -> None: if selection.state in {"preparing", "waiting"}: self._render_candidate_selection_snapshot(selection) diff --git a/src/iac_code/ui/renderer.py b/src/iac_code/ui/renderer.py index 5f565e23..a5f584ab 100644 --- a/src/iac_code/ui/renderer.py +++ b/src/iac_code/ui/renderer.py @@ -1511,9 +1511,17 @@ async def _rebuild_after_transcript(): self._quiet_stop_live(live) live = None spinner = None - # Print current state to scrollback - self._print_segments_to_scrollback(segments, text_buffer) + flush_segments: list[_Segment] = [] + pending_tool_segments: list[_Segment] = [] + for segment in segments: + if segment.kind == "tool" and segment.tool is not None and not segment.tool.done: + pending_tool_segments.append(segment) + else: + flush_segments.append(segment) + if flush_segments or text_buffer: + self._print_segments_to_scrollback(flush_segments, text_buffer) segments.clear() + segments.extend(pending_tool_segments) text_buffer = "" # Handle permission allowed = await permission_handler(event) @@ -1522,6 +1530,21 @@ async def _rebuild_after_transcript(): # defensive pattern in agent_tool / acp/session. if event.response_future is not None and not event.response_future.done(): event.response_future.set_result(allowed) + if segments: + _ensure_live() + _update_live() + if live is not None: + refresh_task = asyncio.create_task( + self._refresh_loop( + live, + segments, + spinner, + lambda: text_buffer, + lambda: task_spinner, + lambda: thinking_buffer, + live_header=live_header, + ) + ) # ── User question request ───────────────────── elif isinstance(event, AskUserQuestionEvent): diff --git a/src/iac_code/ui/repl.py b/src/iac_code/ui/repl.py index 965f0fef..b671e187 100644 --- a/src/iac_code/ui/repl.py +++ b/src/iac_code/ui/repl.py @@ -77,6 +77,7 @@ CandidateDetailEvent, DiagramEvent, PermissionRequestEvent, + StackProgressEvent, SubPipelineStreamEvent, TextDeltaEvent, ToolInputDeltaEvent, @@ -2971,7 +2972,7 @@ def _record_pipeline_display_tool_use( step_id: str | None = None, sub_pipeline_id: str | None = None, ) -> None: - if getattr(event, "name", "") != "complete_step": + if getattr(event, "name", "") not in {"complete_step", "ros_deploy", "ros_stack", "ros_stack_instances"}: return recorder = getattr(self, "_pipeline_display_recorder", None) if recorder is None: @@ -2985,6 +2986,18 @@ def _record_pipeline_display_tool_use( except Exception as exc: logger.warning("Failed to record pipeline display tool use: {}", exc) + def _record_pipeline_display_stack_progress(self, event, *, step_id: str | None = None) -> None: + recorder = getattr(self, "_pipeline_display_recorder", None) + if recorder is None: + return + try: + recorder.record_stack_progress( + event, + step_id=step_id or getattr(self, "_pipeline_display_current_step_id", None), + ) + except Exception as exc: + logger.warning("Failed to record pipeline display stack progress: {}", exc) + def _record_pipeline_display_candidate_diagram(self, event, *, step_id: str | None = None) -> None: recorder = getattr(self, "_pipeline_display_recorder", None) if recorder is None: @@ -4223,6 +4236,8 @@ async def _stop_renderer() -> bool: else: if isinstance(event, ToolUseStartEvent): self._record_pipeline_display_tool_use(event) + if isinstance(event, StackProgressEvent): + self._record_pipeline_display_stack_progress(event) if renderer_task is not None and not renderer_task.done() and agent_events_queue is not None: await agent_events_queue.put(event) else: diff --git a/tests/pipeline/engine/test_display_replay.py b/tests/pipeline/engine/test_display_replay.py index 5aca6c5d..47340c3d 100644 --- a/tests/pipeline/engine/test_display_replay.py +++ b/tests/pipeline/engine/test_display_replay.py @@ -52,6 +52,39 @@ def test_display_replay_ignores_pipeline_warning_without_terminal_change(tmp_pat assert model.attempts[-1].status == "running" +def test_reducer_tracks_latest_stack_progress_for_attempt(tmp_path) -> None: + path = tmp_path / "display.jsonl" + recorder = PipelineDisplayRecorder(path) + + recorder.record("step_started", step_id="deploying", payload={"index": 1, "total": 1}) + recorder.record("tool_used", step_id="deploying", payload={"name": "ros_deploy", "tool_use_id": "tu_deploy"}) + recorder.record( + "stack_progress", + step_id="deploying", + payload={ + "stack_id": "stack-123", + "stack_name": "demo", + "status": "CREATE_IN_PROGRESS", + "progress_percentage": 50.0, + "elapsed_seconds": 30, + "resources": [ + { + "name": "EcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_IN_PROGRESS", + } + ], + }, + ) + + model = PipelineDisplayReducer().reduce(load_display_events(path)) + attempt = model.attempts[-1] + + assert attempt.stack_progress is not None + assert attempt.stack_progress.stack_name == "demo" + assert attempt.stack_progress.resources[0]["name"] == "EcsInstance" + + def test_reducer_attaches_transcript_ids_from_event_payload_and_attempt_metadata(tmp_path): path = tmp_path / "display.jsonl" recorder = PipelineDisplayRecorder(path) diff --git a/tests/ui/test_pipeline_display_replay_renderer.py b/tests/ui/test_pipeline_display_replay_renderer.py index 3c552373..21f0a4df 100644 --- a/tests/ui/test_pipeline_display_replay_renderer.py +++ b/tests/ui/test_pipeline_display_replay_renderer.py @@ -1,4 +1,5 @@ import json +from types import SimpleNamespace from rich.console import Console from rich.text import Text @@ -119,6 +120,38 @@ def test_renderer_prints_generic_pipeline_history(): assert "Interrupted" in text +def test_renderer_prints_stack_progress_snapshot(): + attempt = DisplayAttempt( + step_id="deploying", + attempt_no=1, + index=5, + total=5, + status="running", + tools=[DisplayToolUse(name="ros_deploy", tool_use_id="tu_deploy")], + ) + attempt.stack_progress = SimpleNamespace( + stack_id="stack-123", + stack_name="demo", + status="CREATE_IN_PROGRESS", + progress_percentage=50.0, + elapsed_seconds=30, + resources=[ + { + "name": "EcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_IN_PROGRESS", + } + ], + ) + model = DisplayReplayModel(pipeline_name="selling", attempts=[attempt]) + + text = _render_text(model) + + assert "ROS Deploy" in text + assert "demo" in text + assert "EcsInstance" in text + + def test_renderer_prints_candidate_selection_waiting_state_with_candidate_selection_ui(): model = DisplayReplayModel( pipeline_name="selling", diff --git a/tests/ui/test_renderer_events.py b/tests/ui/test_renderer_events.py index e789d9f5..6790f75b 100644 --- a/tests/ui/test_renderer_events.py +++ b/tests/ui/test_renderer_events.py @@ -1,7 +1,9 @@ import asyncio +from io import StringIO from unittest.mock import MagicMock import pytest +from rich.console import Console from iac_code.services.telemetry.names import Events from iac_code.tools.base import ToolRegistry @@ -14,6 +16,7 @@ MessageEndEvent, MessageStartEvent, PermissionRequestEvent, + StackProgressEvent, TaskNotificationEvent, TextDeltaEvent, ThinkingDeltaEvent, @@ -126,6 +129,76 @@ async def permission_handler(event): assert future.result() is allowed assert legacy_event not in [event_name for event_name, _payload in logged_events] + async def test_stack_progress_remains_visible_after_permission_prompt(self, monkeypatch): + updates = [] + + async def idle_key_listener(*_args, **_kwargs): + await asyncio.Event().wait() + + def fake_live_factory(*args, **kwargs): + live = MagicMock() + live._started = False + live.start = MagicMock() + live.stop = MagicMock() + + def record_update(renderable, *_update_args, **_update_kwargs): + updates.append(renderable) + + live.update.side_effect = record_update + return live + + def render_plain(renderable) -> str: + console = Console(file=StringIO(), force_terminal=True, width=120, color_system=None, _environ={}) + console.print(renderable) + return console.file.getvalue() + + future = asyncio.get_running_loop().create_future() + permission_event = PermissionRequestEvent( + tool_name="ros_deploy", + tool_input={"action": "create"}, + tool_use_id="toolu-deploy", + response_future=future, + ) + + async def events(): + yield MessageStartEvent(message_id="m1") + yield ToolUseStartEvent(tool_use_id="toolu-deploy", name="ros_deploy") + yield ToolUseEndEvent(tool_use_id="toolu-deploy", name="ros_deploy", input={"action": "create"}) + yield MessageEndEvent(stop_reason="tool_use", usage=Usage()) + yield permission_event + yield StackProgressEvent( + stack_id="stack-123", + stack_name="demo", + status="CREATE_IN_PROGRESS", + progress_percentage=50.0, + resources=[ + { + "name": "EcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_IN_PROGRESS", + } + ], + elapsed_seconds=30, + ) + yield ToolResultEvent( + tool_use_id="toolu-deploy", + tool_name="ros_deploy", + result='{"stack_id": "stack-123", "status": "CREATE_COMPLETE", "is_success": true}', + ) + + monkeypatch.setattr(Renderer, "_key_listener", idle_key_listener) + monkeypatch.setattr("iac_code.ui.renderer.Live", fake_live_factory) + renderer = Renderer( + Console(file=StringIO(), force_terminal=True, width=120, color_system=None, _environ={}), + ToolRegistry(), + status_callback=lambda: "test", + ) + + await renderer.run_streaming_output(events(), permission_handler=lambda _event: asyncio.sleep(0, result=True)) + + assert future.result() is True + assert any("EcsInstance" in render_plain(update) for update in updates) + def _make_renderer_for_thinking_test(): from io import StringIO diff --git a/tests/ui/test_repl_pipeline_display_replay.py b/tests/ui/test_repl_pipeline_display_replay.py index 7f9cfa9b..f05c88ba 100644 --- a/tests/ui/test_repl_pipeline_display_replay.py +++ b/tests/ui/test_repl_pipeline_display_replay.py @@ -16,7 +16,7 @@ load_display_events, ) from iac_code.pipeline.engine.events import PipelineEvent, PipelineEventType -from iac_code.types.stream_events import CandidateDetailEvent, DiagramEvent, ToolUseStartEvent +from iac_code.types.stream_events import CandidateDetailEvent, DiagramEvent, StackProgressEvent, ToolUseStartEvent async def _drain_streaming_output(events_iter, **_kwargs): @@ -73,6 +73,49 @@ async def _simple_pipeline_stream(): ) +async def _deploying_pipeline_stream(): + yield PipelineEvent( + type=PipelineEventType.PIPELINE_STARTED, + step_id=None, + timestamp=time.time(), + data={"pipeline_type": "selling", "step_names": ["deploying"]}, + ) + yield PipelineEvent( + type=PipelineEventType.STEP_STARTED, + step_id="deploying", + timestamp=time.time(), + data={"index": 1, "total": 1, "name": "deploying"}, + ) + yield ToolUseStartEvent(tool_use_id="tu_deploy", name="ros_deploy") + yield StackProgressEvent( + stack_id="stack-123", + stack_name="demo", + status="CREATE_IN_PROGRESS", + progress_percentage=50.0, + resources=[ + { + "name": "EcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_IN_PROGRESS", + } + ], + elapsed_seconds=30, + ) + yield ToolUseStartEvent(tool_use_id="tu_complete", name="complete_step") + yield PipelineEvent( + type=PipelineEventType.STEP_COMPLETED, + step_id="deploying", + timestamp=time.time(), + data={"duration_s": 1.0}, + ) + yield PipelineEvent( + type=PipelineEventType.PIPELINE_COMPLETED, + step_id=None, + timestamp=time.time(), + data={"total_steps": 1}, + ) + + @pytest.mark.asyncio async def test_pipeline_stream_records_display_transcript(tmp_path): repl = _make_repl_with_display_recorder(tmp_path) @@ -92,6 +135,28 @@ async def test_pipeline_stream_records_display_transcript(tmp_path): assert events[2]["payload"]["name"] == "complete_step" +@pytest.mark.asyncio +async def test_pipeline_stream_records_deploy_tool_and_stack_progress(tmp_path): + repl = _make_repl_with_display_recorder(tmp_path) + + await repl._render_pipeline_stream(_deploying_pipeline_stream()) + + events = load_display_events(tmp_path / "pipeline" / "display.jsonl") + event_types = [event["type"] for event in events] + assert event_types == [ + "pipeline_started", + "step_started", + "tool_used", + "stack_progress", + "tool_used", + "step_completed", + "pipeline_completed", + ] + assert events[2]["payload"]["name"] == "ros_deploy" + assert events[3]["payload"]["stack_name"] == "demo" + assert events[3]["payload"]["resources"][0]["name"] == "EcsInstance" + + def test_user_aborted_pipeline_records_display_terminal_event(tmp_path): from iac_code.ui.repl import InlineREPL From 474e54c53b4520d094a3d5227a4f51fdab094932 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=A1=82=E9=A9=AC?= Date: Wed, 8 Jul 2026 23:07:01 +0800 Subject: [PATCH 2/3] Tighten ROS deploy progress replay coverage --- .../pipeline/engine/display_replay.py | 16 ++++- src/iac_code/ui/pipeline_display_replay.py | 4 +- src/iac_code/ui/renderer.py | 2 + src/iac_code/ui/repl.py | 2 +- tests/pipeline/engine/test_display_replay.py | 38 ++++++++-- .../selling/tools/test_ros_deploy_tool.py | 72 +++++++++++++++++++ .../test_pipeline_display_replay_renderer.py | 39 ++++++---- tests/ui/test_renderer_events.py | 13 +++- 8 files changed, 161 insertions(+), 25 deletions(-) diff --git a/src/iac_code/pipeline/engine/display_replay.py b/src/iac_code/pipeline/engine/display_replay.py index 9c4a3677..957a5018 100644 --- a/src/iac_code/pipeline/engine/display_replay.py +++ b/src/iac_code/pipeline/engine/display_replay.py @@ -93,7 +93,7 @@ class DisplayAttempt: rollback_reason: str = "" error: str = "" tools: list[DisplayToolUse] = field(default_factory=list) - stack_progress: DisplayStackProgress | None = None + stack_progresses: list[DisplayStackProgress] = field(default_factory=list) sub_pipelines: dict[str, DisplaySubPipeline] = field(default_factory=dict) candidate_selection: DisplayCandidateSelection = field(default_factory=DisplayCandidateSelection) @@ -342,7 +342,7 @@ def reduce( if event_type == "stack_progress" and attempt is not None: resources = payload.get("resources") - attempt.stack_progress = DisplayStackProgress( + progress = DisplayStackProgress( stack_id=str(payload.get("stack_id") or ""), stack_name=str(payload.get("stack_name") or ""), status=str(payload.get("status") or ""), @@ -354,6 +354,18 @@ def reduce( if isinstance(resource, dict) ], ) + existing_index = next( + ( + index + for index, item in enumerate(attempt.stack_progresses) + if item.stack_id and item.stack_id == progress.stack_id + ), + None, + ) + if existing_index is None: + attempt.stack_progresses.append(progress) + else: + attempt.stack_progresses[existing_index] = progress continue if event_type == "user_input_required" and attempt is not None: diff --git a/src/iac_code/ui/pipeline_display_replay.py b/src/iac_code/ui/pipeline_display_replay.py index 8b590641..1184fc31 100644 --- a/src/iac_code/ui/pipeline_display_replay.py +++ b/src/iac_code/ui/pipeline_display_replay.py @@ -76,8 +76,8 @@ def _render_attempt(self, attempt: DisplayAttempt) -> None: for tool in attempt.tools: self.console.print(f" ● {display_tool_name(tool.name)}") - if attempt.stack_progress is not None: - self._render_stack_progress(attempt.stack_progress) + for progress in attempt.stack_progresses: + self._render_stack_progress(progress) if attempt.step_type == "parallel_sub_pipeline" or attempt.sub_pipelines: self._render_sub_pipelines(attempt) diff --git a/src/iac_code/ui/renderer.py b/src/iac_code/ui/renderer.py index a5f584ab..5ca3cf7a 100644 --- a/src/iac_code/ui/renderer.py +++ b/src/iac_code/ui/renderer.py @@ -1545,6 +1545,8 @@ async def _rebuild_after_transcript(): live_header=live_header, ) ) + key_task = _new_key_task() + await asyncio.sleep(0) # ── User question request ───────────────────── elif isinstance(event, AskUserQuestionEvent): diff --git a/src/iac_code/ui/repl.py b/src/iac_code/ui/repl.py index b671e187..308a9f61 100644 --- a/src/iac_code/ui/repl.py +++ b/src/iac_code/ui/repl.py @@ -2972,7 +2972,7 @@ def _record_pipeline_display_tool_use( step_id: str | None = None, sub_pipeline_id: str | None = None, ) -> None: - if getattr(event, "name", "") not in {"complete_step", "ros_deploy", "ros_stack", "ros_stack_instances"}: + if getattr(event, "name", "") not in {"complete_step", "ros_deploy", "ros_stack"}: return recorder = getattr(self, "_pipeline_display_recorder", None) if recorder is None: diff --git a/tests/pipeline/engine/test_display_replay.py b/tests/pipeline/engine/test_display_replay.py index 47340c3d..acf683b3 100644 --- a/tests/pipeline/engine/test_display_replay.py +++ b/tests/pipeline/engine/test_display_replay.py @@ -52,7 +52,7 @@ def test_display_replay_ignores_pipeline_warning_without_terminal_change(tmp_pat assert model.attempts[-1].status == "running" -def test_reducer_tracks_latest_stack_progress_for_attempt(tmp_path) -> None: +def test_reducer_tracks_stack_progress_snapshots_per_stack_for_attempt(tmp_path) -> None: path = tmp_path / "display.jsonl" recorder = PipelineDisplayRecorder(path) @@ -76,13 +76,43 @@ def test_reducer_tracks_latest_stack_progress_for_attempt(tmp_path) -> None: ], }, ) + recorder.record( + "stack_progress", + step_id="deploying", + payload={ + "stack_id": "stack-new", + "stack_name": "demo-replacement", + "status": "CREATE_IN_PROGRESS", + "progress_percentage": 25.0, + "elapsed_seconds": 45, + "resources": [ + { + "name": "ReplacementEcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_IN_PROGRESS", + } + ], + }, + ) + recorder.record( + "stack_progress", + step_id="deploying", + payload={ + "stack_id": "stack-123", + "stack_name": "demo", + "status": "DELETE_COMPLETE", + "progress_percentage": 100.0, + "elapsed_seconds": 60, + "resources": [], + }, + ) model = PipelineDisplayReducer().reduce(load_display_events(path)) attempt = model.attempts[-1] - assert attempt.stack_progress is not None - assert attempt.stack_progress.stack_name == "demo" - assert attempt.stack_progress.resources[0]["name"] == "EcsInstance" + assert [progress.stack_id for progress in attempt.stack_progresses] == ["stack-123", "stack-new"] + assert attempt.stack_progresses[0].status == "DELETE_COMPLETE" + assert attempt.stack_progresses[1].resources[0]["name"] == "ReplacementEcsInstance" def test_reducer_attaches_transcript_ids_from_event_payload_and_attempt_metadata(tmp_path): diff --git a/tests/pipeline/selling/tools/test_ros_deploy_tool.py b/tests/pipeline/selling/tools/test_ros_deploy_tool.py index 7b8a2b30..495a8cdc 100644 --- a/tests/pipeline/selling/tools/test_ros_deploy_tool.py +++ b/tests/pipeline/selling/tools/test_ros_deploy_tool.py @@ -1,3 +1,4 @@ +import asyncio import json import os @@ -339,6 +340,77 @@ async def test_wait_action_rejects_create_only_fields_without_polling(monkeypatc assert fake_stack.wait_calls == [] +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("tool_input", "guard_state", "expected_call_count", "uses_wait"), + [ + ( + {"action": "create", "stack_name": "demo", "template_url": "templates/demo.yml"}, + {}, + 1, + False, + ), + ( + {"action": "continue_create", "stack_id": "stack-failed", "template_url": "templates/demo.yml"}, + {"ros_deploy_owned_stack_ids": {"stack-failed": {"action": "create"}}}, + 1, + False, + ), + ( + {"action": "wait", "stack_id": "stack-slow", "region_id": "cn-hangzhou"}, + {}, + 1, + True, + ), + ( + { + "action": "delete_and_create", + "stack_id": "stack-old", + "stack_name": "demo", + "template_url": "templates/demo.yml", + }, + {"ros_deploy_owned_stack_ids": {"stack-old": {"action": "create"}}}, + 2, + False, + ), + ], +) +async def test_ros_deploy_preserves_event_queue_for_all_progress_actions( + monkeypatch, + tmp_path, + tool_input, + guard_state, + expected_call_count, + uses_wait, +): + templates_dir = tmp_path / "templates" + templates_dir.mkdir() + (templates_dir / "demo.yml").write_text("ROSTemplateFormatVersion: '2015-09-01'\n", encoding="utf-8") + results = [ + ToolResult.success( + json.dumps( + { + "stack_id": tool_input.get("stack_id") if index == 0 else "stack-new", + "stack_name": "demo", + "status": "CREATE_COMPLETE", + "is_success": True, + } + ) + ) + for index in range(expected_call_count) + ] + tool, fake_stack = _deploy_tool(monkeypatch, guard_state=guard_state, results=results) + event_queue: asyncio.Queue = asyncio.Queue() + context = ToolContext(cwd=str(tmp_path), pipeline_mode=True, event_queue=event_queue) + + result = await tool.execute(tool_input=tool_input, context=context) + + assert result.is_error is False + contexts = [call[-1] for call in (fake_stack.wait_calls if uses_wait else fake_stack.calls)] + assert len(contexts) == expected_call_count + assert all(call_context.event_queue is event_queue for call_context in contexts) + + @pytest.mark.asyncio async def test_continue_validation_failure_recommends_delete_and_create_without_running_it(monkeypatch): guard_state = {"ros_deploy_owned_stack_ids": {"stack-failed": {"action": "create"}}} diff --git a/tests/ui/test_pipeline_display_replay_renderer.py b/tests/ui/test_pipeline_display_replay_renderer.py index 21f0a4df..3ee7e19f 100644 --- a/tests/ui/test_pipeline_display_replay_renderer.py +++ b/tests/ui/test_pipeline_display_replay_renderer.py @@ -129,25 +129,38 @@ def test_renderer_prints_stack_progress_snapshot(): status="running", tools=[DisplayToolUse(name="ros_deploy", tool_use_id="tu_deploy")], ) - attempt.stack_progress = SimpleNamespace( - stack_id="stack-123", - stack_name="demo", - status="CREATE_IN_PROGRESS", - progress_percentage=50.0, - elapsed_seconds=30, - resources=[ - { - "name": "EcsInstance", - "resource_type": "ALIYUN::ECS::Instance", - "status": "CREATE_IN_PROGRESS", - } - ], + attempt.stack_progresses.extend( + [ + SimpleNamespace( + stack_id="stack-old", + stack_name="demo-old", + status="DELETE_COMPLETE", + progress_percentage=100.0, + elapsed_seconds=30, + resources=[], + ), + SimpleNamespace( + stack_id="stack-new", + stack_name="demo", + status="CREATE_IN_PROGRESS", + progress_percentage=50.0, + elapsed_seconds=45, + resources=[ + { + "name": "EcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_IN_PROGRESS", + } + ], + ), + ] ) model = DisplayReplayModel(pipeline_name="selling", attempts=[attempt]) text = _render_text(model) assert "ROS Deploy" in text + assert "demo-old" in text assert "demo" in text assert "EcsInstance" in text diff --git a/tests/ui/test_renderer_events.py b/tests/ui/test_renderer_events.py index 6790f75b..86812d85 100644 --- a/tests/ui/test_renderer_events.py +++ b/tests/ui/test_renderer_events.py @@ -131,9 +131,16 @@ async def permission_handler(event): async def test_stack_progress_remains_visible_after_permission_prompt(self, monkeypatch): updates = [] + key_listener_calls = 0 - async def idle_key_listener(*_args, **_kwargs): - await asyncio.Event().wait() + def idle_key_listener(*_args, **_kwargs): + nonlocal key_listener_calls + key_listener_calls += 1 + + async def wait_forever(): + await asyncio.Event().wait() + + return wait_forever() def fake_live_factory(*args, **kwargs): live = MagicMock() @@ -161,7 +168,6 @@ def render_plain(renderable) -> str: ) async def events(): - yield MessageStartEvent(message_id="m1") yield ToolUseStartEvent(tool_use_id="toolu-deploy", name="ros_deploy") yield ToolUseEndEvent(tool_use_id="toolu-deploy", name="ros_deploy", input={"action": "create"}) yield MessageEndEvent(stop_reason="tool_use", usage=Usage()) @@ -197,6 +203,7 @@ async def events(): await renderer.run_streaming_output(events(), permission_handler=lambda _event: asyncio.sleep(0, result=True)) assert future.result() is True + assert key_listener_calls >= 2 assert any("EcsInstance" in render_plain(update) for update in updates) From 5500aa2ef90250b1f6973dc639e6a64bfe87bfc9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=A1=82=E9=A9=AC?= Date: Wed, 8 Jul 2026 23:42:16 +0800 Subject: [PATCH 3/3] Coalesce ROS deploy replay progress snapshots --- .../i18n/locales/de/LC_MESSAGES/messages.po | 5 ++ .../i18n/locales/es/LC_MESSAGES/messages.po | 5 ++ .../i18n/locales/fr/LC_MESSAGES/messages.po | 5 ++ .../i18n/locales/ja/LC_MESSAGES/messages.po | 5 ++ .../i18n/locales/pt/LC_MESSAGES/messages.po | 5 ++ .../i18n/locales/zh/LC_MESSAGES/messages.po | 5 ++ .../pipeline/engine/display_replay.py | 87 +++++++++++++++++-- src/iac_code/ui/pipeline_display_replay.py | 8 +- tests/pipeline/engine/test_display_replay.py | 46 ++++++++++ .../selling/tools/test_ros_deploy_tool.py | 56 +++++++++--- .../test_pipeline_display_replay_renderer.py | 1 + 11 files changed, 205 insertions(+), 23 deletions(-) diff --git a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po index 96513837..9aa3863c 100644 --- a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po @@ -5300,6 +5300,11 @@ msgstr " ✓ {name}: abgeschlossen" msgid "Rolled back" msgstr "Zurückgesetzt" +#: src/iac_code/ui/pipeline_display_replay.py +#, python-brace-format +msgid " Last observed stack: {label} [{status}] {progress:.0f}%" +msgstr " Zuletzt beobachteter Stack: {label} [{status}] {progress:.0f}%" + #: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py msgid "Resource" msgstr "Ressource" diff --git a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po index 92a3d2ee..ca6ab232 100644 --- a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po @@ -5286,6 +5286,11 @@ msgstr " ✓ {name}: completado" msgid "Rolled back" msgstr "Revertido" +#: src/iac_code/ui/pipeline_display_replay.py +#, python-brace-format +msgid " Last observed stack: {label} [{status}] {progress:.0f}%" +msgstr " Última pila observada: {label} [{status}] {progress:.0f}%" + #: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py msgid "Resource" msgstr "Recurso" diff --git a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po index 894b6cd9..7852fa32 100644 --- a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po @@ -5305,6 +5305,11 @@ msgstr " ✓ {name} : terminé" msgid "Rolled back" msgstr "Revenu en arrière" +#: src/iac_code/ui/pipeline_display_replay.py +#, python-brace-format +msgid " Last observed stack: {label} [{status}] {progress:.0f}%" +msgstr " Dernière pile observée : {label} [{status}] {progress:.0f}%" + #: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py msgid "Resource" msgstr "Ressource" diff --git a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po index b3b16fbd..df10de54 100644 --- a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po @@ -5035,6 +5035,11 @@ msgstr " ✓ {name}: 完了" msgid "Rolled back" msgstr "ロールバック済み" +#: src/iac_code/ui/pipeline_display_replay.py +#, python-brace-format +msgid " Last observed stack: {label} [{status}] {progress:.0f}%" +msgstr " 最後に観測したスタック: {label} [{status}] {progress:.0f}%" + #: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py msgid "Resource" msgstr "リソース" diff --git a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po index e67ab9bd..6d9768d9 100644 --- a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po @@ -5241,6 +5241,11 @@ msgstr " ✓ {name}: concluído" msgid "Rolled back" msgstr "Revertido" +#: src/iac_code/ui/pipeline_display_replay.py +#, python-brace-format +msgid " Last observed stack: {label} [{status}] {progress:.0f}%" +msgstr " Última pilha observada: {label} [{status}] {progress:.0f}%" + #: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py msgid "Resource" msgstr "Recurso" diff --git a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po index 7dc81367..d2ec0ef4 100644 --- a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po +++ b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po @@ -4966,6 +4966,11 @@ msgstr " ✓ {name}: 已完成" msgid "Rolled back" msgstr "已回滚" +#: src/iac_code/ui/pipeline_display_replay.py +#, python-brace-format +msgid " Last observed stack: {label} [{status}] {progress:.0f}%" +msgstr " 最后观测到的栈: {label} [{status}] {progress:.0f}%" + #: src/iac_code/ui/pipeline_display_replay.py src/iac_code/ui/renderer.py msgid "Resource" msgstr "资源" diff --git a/src/iac_code/pipeline/engine/display_replay.py b/src/iac_code/pipeline/engine/display_replay.py index 957a5018..c0010a0c 100644 --- a/src/iac_code/pipeline/engine/display_replay.py +++ b/src/iac_code/pipeline/engine/display_replay.py @@ -14,6 +14,14 @@ DISPLAY_TRANSCRIPT_FILENAME = "display.jsonl" TERMINAL_ATTEMPT_STATUSES = {"completed", "failed", "rolled_back", "interrupted"} +STACK_PROGRESS_RECORD_DELTA = 5.0 + + +@dataclass(frozen=True) +class _StackProgressRecordSnapshot: + status: str + progress_percentage: float + resources_signature: tuple[tuple[str, str, str, str], ...] @dataclass @@ -113,6 +121,7 @@ class PipelineDisplayRecorder: def __init__(self, path: str | Path | None) -> None: self.path = Path(path) if path is not None else None + self._stack_progress_snapshots: dict[tuple[str, str], _StackProgressRecordSnapshot] = {} @property def enabled(self) -> bool: @@ -165,18 +174,80 @@ def record_tool_use(self, event: Any, *, step_id: str | None = None, sub_pipelin def record_stack_progress(self, event: Any, *, step_id: str | None = None) -> None: resources = getattr(event, "resources", []) + payload = { + "stack_id": getattr(event, "stack_id", ""), + "stack_name": getattr(event, "stack_name", ""), + "status": getattr(event, "status", ""), + "progress_percentage": getattr(event, "progress_percentage", 0.0), + "elapsed_seconds": getattr(event, "elapsed_seconds", 0), + "resources": resources if isinstance(resources, list) else [], + } + if not self._should_record_stack_progress(step_id, payload): + return self.record( "stack_progress", step_id=step_id, - payload={ - "stack_id": getattr(event, "stack_id", ""), - "stack_name": getattr(event, "stack_name", ""), - "status": getattr(event, "status", ""), - "progress_percentage": getattr(event, "progress_percentage", 0.0), - "elapsed_seconds": getattr(event, "elapsed_seconds", 0), - "resources": resources if isinstance(resources, list) else [], - }, + payload=payload, + ) + + def _should_record_stack_progress(self, step_id: str | None, payload: dict[str, Any]) -> bool: + key = self._stack_progress_record_key(step_id, payload) + if key is None: + return True + + current = _StackProgressRecordSnapshot( + status=str(payload.get("status") or ""), + progress_percentage=self._payload_float(payload.get("progress_percentage")), + resources_signature=self._resources_signature(payload.get("resources")), + ) + previous = self._stack_progress_snapshots.get(key) + if previous is None: + self._stack_progress_snapshots[key] = current + return True + + should_record = ( + current.status != previous.status + or current.resources_signature != previous.resources_signature + or abs(current.progress_percentage - previous.progress_percentage) >= STACK_PROGRESS_RECORD_DELTA ) + if should_record: + self._stack_progress_snapshots[key] = current + return should_record + + @staticmethod + def _stack_progress_record_key(step_id: str | None, payload: dict[str, Any]) -> tuple[str, str] | None: + stack_key = str(payload.get("stack_id") or "") + if not stack_key: + stack_name = str(payload.get("stack_name") or "") + stack_key = f"name:{stack_name}" if stack_name else "" + if not stack_key: + return None + return (step_id or "", stack_key) + + @staticmethod + def _payload_float(value: Any) -> float: + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + + @staticmethod + def _resources_signature(resources: Any) -> tuple[tuple[str, str, str, str], ...]: + if not isinstance(resources, list): + return () + signature = [] + for resource in resources: + if not isinstance(resource, dict): + continue + signature.append( + ( + str(resource.get("name") or ""), + str(resource.get("resource_type") or ""), + str(resource.get("status") or ""), + str(resource.get("status_reason") or ""), + ) + ) + return tuple(sorted(signature)) def record_candidate_diagram(self, event: Any, *, step_id: str | None = None) -> None: self.record( diff --git a/src/iac_code/ui/pipeline_display_replay.py b/src/iac_code/ui/pipeline_display_replay.py index 1184fc31..f2eda272 100644 --- a/src/iac_code/ui/pipeline_display_replay.py +++ b/src/iac_code/ui/pipeline_display_replay.py @@ -206,7 +206,13 @@ def _render_stack_progress(self, progress: Any) -> None: if stack_name and stack_id: label = f"{stack_name}({stack_id})" if label: - self.console.print(f" Stack: {label} [{stack_status}] {progress_percentage:.0f}%") + self.console.print( + _(" Last observed stack: {label} [{status}] {progress:.0f}%").format( + label=label, + status=stack_status, + progress=progress_percentage, + ) + ) resources = getattr(progress, "resources", []) if not isinstance(resources, list) or not resources: diff --git a/tests/pipeline/engine/test_display_replay.py b/tests/pipeline/engine/test_display_replay.py index acf683b3..0d2c9ab6 100644 --- a/tests/pipeline/engine/test_display_replay.py +++ b/tests/pipeline/engine/test_display_replay.py @@ -1,4 +1,5 @@ import json +from types import SimpleNamespace from iac_code.pipeline.engine.display_replay import ( PipelineDisplayRecorder, @@ -115,6 +116,51 @@ def test_reducer_tracks_stack_progress_snapshots_per_stack_for_attempt(tmp_path) assert attempt.stack_progresses[1].resources[0]["name"] == "ReplacementEcsInstance" +def test_recorder_coalesces_repeated_stack_progress_snapshots(tmp_path) -> None: + path = tmp_path / "display.jsonl" + recorder = PipelineDisplayRecorder(path) + + in_progress_resource = [ + { + "name": "EcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_IN_PROGRESS", + "status_reason": "", + } + ] + complete_resource = [ + { + "name": "EcsInstance", + "resource_type": "ALIYUN::ECS::Instance", + "status": "CREATE_COMPLETE", + "status_reason": "", + } + ] + + def event(status: str, progress: float, resources: list[dict]): + return SimpleNamespace( + stack_id="stack-123", + stack_name="demo", + status=status, + progress_percentage=progress, + elapsed_seconds=30, + resources=resources, + ) + + for _ in range(10): + recorder.record_stack_progress(event("CREATE_IN_PROGRESS", 50.0, in_progress_resource), step_id="deploying") + recorder.record_stack_progress(event("CREATE_IN_PROGRESS", 52.0, in_progress_resource), step_id="deploying") + recorder.record_stack_progress(event("CREATE_IN_PROGRESS", 55.0, in_progress_resource), step_id="deploying") + recorder.record_stack_progress(event("CREATE_IN_PROGRESS", 55.0, complete_resource), step_id="deploying") + recorder.record_stack_progress(event("CREATE_COMPLETE", 100.0, complete_resource), step_id="deploying") + recorder.record_stack_progress(event("CREATE_COMPLETE", 100.0, complete_resource), step_id="deploying") + + events = [item for item in load_display_events(path) if item["type"] == "stack_progress"] + + assert [item["payload"]["progress_percentage"] for item in events] == [50.0, 55.0, 55.0, 100.0] + assert events[-1]["payload"]["status"] == "CREATE_COMPLETE" + + def test_reducer_attaches_transcript_ids_from_event_payload_and_attempt_metadata(tmp_path): path = tmp_path / "display.jsonl" recorder = PipelineDisplayRecorder(path) diff --git a/tests/pipeline/selling/tools/test_ros_deploy_tool.py b/tests/pipeline/selling/tools/test_ros_deploy_tool.py index 495a8cdc..8d6cd6a1 100644 --- a/tests/pipeline/selling/tools/test_ros_deploy_tool.py +++ b/tests/pipeline/selling/tools/test_ros_deploy_tool.py @@ -6,6 +6,7 @@ from iac_code.tools.base import ToolContext, ToolResult from iac_code.types.permissions import PermissionMode, ToolPermissionContext +from iac_code.types.stream_events import StackProgressEvent def _permission_ctx(*, allow=None, deny=None, mode=PermissionMode.DEFAULT): @@ -18,28 +19,35 @@ def _permission_ctx(*, allow=None, deny=None, mode=PermissionMode.DEFAULT): class FakeRosStack: - def __init__(self, results): + def __init__(self, results, progress_events=None): self.results = list(results) + self.progress_events = list(progress_events or []) self.calls = [] self.wait_calls = [] + async def _emit_next_progress(self, context): + if context.event_queue is not None and self.progress_events: + await context.event_queue.put(self.progress_events.pop(0)) + async def execute(self, *, tool_input, context): self.calls.append((tool_input, context)) + await self._emit_next_progress(context) if not self.results: raise AssertionError("unexpected ros stack call") return self.results.pop(0) async def wait_for_stack_operation(self, action, params, region, stack_id, context): self.wait_calls.append((action, params, region, stack_id, context)) + await self._emit_next_progress(context) if not self.results: raise AssertionError("unexpected ros stack wait call") return self.results.pop(0) -def _deploy_tool(monkeypatch, *, guard_state=None, results=None): +def _deploy_tool(monkeypatch, *, guard_state=None, results=None, progress_events=None): from iac_code.pipeline.selling.tools.ros_deploy_tool import RosDeployTool - fake_stack = FakeRosStack(results or []) + fake_stack = FakeRosStack(results or [], progress_events=progress_events) tool = RosDeployTool(completion_guard_state=guard_state if guard_state is not None else {}) monkeypatch.setattr(tool, "_new_stack_tool", lambda: fake_stack) return tool, fake_stack @@ -342,24 +350,24 @@ async def test_wait_action_rejects_create_only_fields_without_polling(monkeypatc @pytest.mark.asyncio @pytest.mark.parametrize( - ("tool_input", "guard_state", "expected_call_count", "uses_wait"), + ("tool_input", "guard_state", "expected_progress_stack_ids", "uses_wait"), [ ( {"action": "create", "stack_name": "demo", "template_url": "templates/demo.yml"}, {}, - 1, + ["stack-created"], False, ), ( {"action": "continue_create", "stack_id": "stack-failed", "template_url": "templates/demo.yml"}, {"ros_deploy_owned_stack_ids": {"stack-failed": {"action": "create"}}}, - 1, + ["stack-failed"], False, ), ( {"action": "wait", "stack_id": "stack-slow", "region_id": "cn-hangzhou"}, {}, - 1, + ["stack-slow"], True, ), ( @@ -370,7 +378,7 @@ async def test_wait_action_rejects_create_only_fields_without_polling(monkeypatc "template_url": "templates/demo.yml", }, {"ros_deploy_owned_stack_ids": {"stack-old": {"action": "create"}}}, - 2, + ["stack-old", "stack-new"], False, ), ], @@ -380,26 +388,42 @@ async def test_ros_deploy_preserves_event_queue_for_all_progress_actions( tmp_path, tool_input, guard_state, - expected_call_count, + expected_progress_stack_ids, uses_wait, ): templates_dir = tmp_path / "templates" templates_dir.mkdir() (templates_dir / "demo.yml").write_text("ROSTemplateFormatVersion: '2015-09-01'\n", encoding="utf-8") + progress_events = [ + StackProgressEvent( + stack_id=stack_id, + stack_name="demo", + status="DELETE_COMPLETE" if stack_id == "stack-old" else "CREATE_COMPLETE", + progress_percentage=100.0, + resources=[], + elapsed_seconds=index + 1, + ) + for index, stack_id in enumerate(expected_progress_stack_ids) + ] results = [ ToolResult.success( json.dumps( { - "stack_id": tool_input.get("stack_id") if index == 0 else "stack-new", + "stack_id": stack_id, "stack_name": "demo", - "status": "CREATE_COMPLETE", + "status": "DELETE_COMPLETE" if stack_id == "stack-old" else "CREATE_COMPLETE", "is_success": True, } ) ) - for index in range(expected_call_count) + for stack_id in expected_progress_stack_ids ] - tool, fake_stack = _deploy_tool(monkeypatch, guard_state=guard_state, results=results) + tool, fake_stack = _deploy_tool( + monkeypatch, + guard_state=guard_state, + results=results, + progress_events=progress_events, + ) event_queue: asyncio.Queue = asyncio.Queue() context = ToolContext(cwd=str(tmp_path), pipeline_mode=True, event_queue=event_queue) @@ -407,8 +431,12 @@ async def test_ros_deploy_preserves_event_queue_for_all_progress_actions( assert result.is_error is False contexts = [call[-1] for call in (fake_stack.wait_calls if uses_wait else fake_stack.calls)] - assert len(contexts) == expected_call_count + assert len(contexts) == len(expected_progress_stack_ids) assert all(call_context.event_queue is event_queue for call_context in contexts) + emitted_events = [] + while not event_queue.empty(): + emitted_events.append(event_queue.get_nowait()) + assert [event.stack_id for event in emitted_events] == expected_progress_stack_ids @pytest.mark.asyncio diff --git a/tests/ui/test_pipeline_display_replay_renderer.py b/tests/ui/test_pipeline_display_replay_renderer.py index 3ee7e19f..170bf8c4 100644 --- a/tests/ui/test_pipeline_display_replay_renderer.py +++ b/tests/ui/test_pipeline_display_replay_renderer.py @@ -160,6 +160,7 @@ def test_renderer_prints_stack_progress_snapshot(): text = _render_text(model) assert "ROS Deploy" in text + assert "Last observed stack:" in text assert "demo-old" in text assert "demo" in text assert "EcsInstance" in text