Fix pipeline tool result rendering

src/iac_code/agent/agent_loop.py

@@ -1276,7 +1276,7 @@ async def poll_event_queues():
                             tool_use_id=req.id,
                             content=processed.content,
                             is_error=result.is_error,
-                            metadata=self._tool_result_block_metadata(processed),
+                            metadata=self._tool_result_block_metadata(processed, result_metadata),
                         )
                     )
 
@@ -1406,10 +1406,30 @@ def _tool_result_render_metadata(
         return cls._merge_tool_render_metadata(metadata, render_metadata)
 
     @staticmethod
-    def _tool_result_block_metadata(processed: Any) -> dict[str, Any]:
-        if not getattr(processed, "is_externalized", False) or not getattr(processed, "file_path", None):
-            return {}
-        return {EXTERNALIZED_RESULT_PATH_METADATA_KEY: str(processed.file_path)}
+    def _tool_result_block_metadata(processed: Any, result_metadata: dict[str, Any] | None = None) -> dict[str, Any]:
+        metadata: dict[str, Any] = {}
+        if getattr(processed, "is_externalized", False) and getattr(processed, "file_path", None):
+            metadata[EXTERNALIZED_RESULT_PATH_METADATA_KEY] = str(processed.file_path)
+
+        if isinstance(result_metadata, dict):
+            render_metadata = result_metadata.get(TOOL_RENDER_METADATA_KEY)
+            if isinstance(render_metadata, dict):
+                safe_render_metadata: dict[str, Any] = {}
+                for key in (
+                    TOOL_RENDER_DISPLAY_NAME_KEY,
+                    TOOL_RENDER_RESULT_COMPACT_KEY,
+                    TOOL_RENDER_RESULT_VERBOSE_KEY,
+                ):
+                    value = render_metadata.get(key)
+                    if isinstance(value, str):
+                        safe_render_metadata[key] = value
+                value = render_metadata.get(TOOL_RENDER_VERBOSE_RESULT_IN_TRANSCRIPT_KEY)
+                if isinstance(value, bool):
+                    safe_render_metadata[TOOL_RENDER_VERBOSE_RESULT_IN_TRANSCRIPT_KEY] = value
+                if safe_render_metadata:
+                    metadata[TOOL_RENDER_METADATA_KEY] = safe_render_metadata
+
+        return metadata
 
     def _mark_read_memory_tool_result(self, request: ToolCallRequest, result: ToolResult) -> None:
         if request.name != "read_memory" or result.is_error or self._memory_recall_service is None:

src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po

@@ -3069,6 +3069,10 @@ msgstr ""
 msgid "The user-facing question to ask."
 msgstr "Die dem Benutzer zu stellende Frage."
 
+#: src/iac_code/pipeline/engine/ask_user_question_tool.py
+msgid "ask_user_question validation failed."
+msgstr "Die Validierung von ask_user_question ist fehlgeschlagen."
+
 #: src/iac_code/pipeline/engine/ask_user_question_tool.py
 msgid "ask_user_question requires a pipeline event queue."
 msgstr "ask_user_question erfordert eine Pipeline-Ereigniswarteschlange."
@@ -3365,6 +3369,11 @@ msgstr "conclusion muss dieser Schema-Zusammenfassung entsprechen:\n"
 msgid "<fill according to the current step requirements>"
 msgstr "<gemäß den Anforderungen des aktuellen Schritts ausfüllen>"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+#: src/iac_code/ui/renderer.py
+msgid "complete_step validation failed."
+msgstr "Die Validierung von complete_step ist fehlgeschlagen."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid "Clarification is required before completing the current step."
 msgstr "Vor Abschluss des aktuellen Schritts ist eine Klärung erforderlich."

src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po

@@ -3070,6 +3070,10 @@ msgstr ""
 msgid "The user-facing question to ask."
 msgstr "La pregunta que se mostrará al usuario."
 
+#: src/iac_code/pipeline/engine/ask_user_question_tool.py
+msgid "ask_user_question validation failed."
+msgstr "La validación de ask_user_question falló."
+
 #: src/iac_code/pipeline/engine/ask_user_question_tool.py
 msgid "ask_user_question requires a pipeline event queue."
 msgstr "ask_user_question requiere una cola de eventos del pipeline."
@@ -3357,6 +3361,11 @@ msgstr "conclusion debe coincidir con este resumen de esquema:\n"
 msgid "<fill according to the current step requirements>"
 msgstr "<rellenar según los requisitos del paso actual>"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+#: src/iac_code/ui/renderer.py
+msgid "complete_step validation failed."
+msgstr "La validación de complete_step falló."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid "Clarification is required before completing the current step."
 msgstr "Se requiere una aclaración antes de completar el paso actual."

src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po

@@ -3077,6 +3077,10 @@ msgstr ""
 msgid "The user-facing question to ask."
 msgstr "Question à poser à l’utilisateur."
 
+#: src/iac_code/pipeline/engine/ask_user_question_tool.py
+msgid "ask_user_question validation failed."
+msgstr "La validation de ask_user_question a échoué."
+
 #: src/iac_code/pipeline/engine/ask_user_question_tool.py
 msgid "ask_user_question requires a pipeline event queue."
 msgstr "ask_user_question nécessite une file d’événements du pipeline."
@@ -3364,6 +3368,11 @@ msgstr "conclusion doit correspondre à ce résumé de schéma :\n"
 msgid "<fill according to the current step requirements>"
 msgstr "<renseigner selon les exigences de l’étape actuelle>"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+#: src/iac_code/ui/renderer.py
+msgid "complete_step validation failed."
+msgstr "La validation de complete_step a échoué."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid "Clarification is required before completing the current step."
 msgstr "Une clarification est requise avant de terminer l’étape actuelle."

src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po

@@ -2958,6 +2958,10 @@ msgstr "ユーザーに選択肢の選択または確認内容の入力を求め
 msgid "The user-facing question to ask."
 msgstr "ユーザーに表示する質問です。"
 
+#: src/iac_code/pipeline/engine/ask_user_question_tool.py
+msgid "ask_user_question validation failed."
+msgstr "ask_user_question の検証に失敗しました。"
+
 #: src/iac_code/pipeline/engine/ask_user_question_tool.py
 msgid "ask_user_question requires a pipeline event queue."
 msgstr "ask_user_question にはパイプラインイベントキューが必要です。"
@@ -3200,6 +3204,11 @@ msgstr "conclusion は次のスキーマ概要に一致する必要がありま
 msgid "<fill according to the current step requirements>"
 msgstr "<現在のステップ要件に従って入力>"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+#: src/iac_code/ui/renderer.py
+msgid "complete_step validation failed."
+msgstr "complete_step の検証に失敗しました。"
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid "Clarification is required before completing the current step."
 msgstr "現在のステップを完了する前に確認が必要です。"

src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po

@@ -3053,6 +3053,10 @@ msgstr ""
 msgid "The user-facing question to ask."
 msgstr "A pergunta a ser exibida ao usuário."
 
+#: src/iac_code/pipeline/engine/ask_user_question_tool.py
+msgid "ask_user_question validation failed."
+msgstr "A validação de ask_user_question falhou."
+
 #: src/iac_code/pipeline/engine/ask_user_question_tool.py
 msgid "ask_user_question requires a pipeline event queue."
 msgstr "ask_user_question requer uma fila de eventos do pipeline."
@@ -3334,6 +3338,11 @@ msgstr "conclusion deve corresponder a este resumo de esquema:\n"
 msgid "<fill according to the current step requirements>"
 msgstr "<preencher conforme os requisitos da etapa atual>"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+#: src/iac_code/ui/renderer.py
+msgid "complete_step validation failed."
+msgstr "A validação de complete_step falhou."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid "Clarification is required before completing the current step."
 msgstr "É necessário esclarecimento antes de concluir a etapa atual."

src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po

@@ -2934,6 +2934,10 @@ msgstr "仅用于 pipeline 的工具，用于请用户选择选项或输入澄
 msgid "The user-facing question to ask."
 msgstr "要询问用户的问题。"
 
+#: src/iac_code/pipeline/engine/ask_user_question_tool.py
+msgid "ask_user_question validation failed."
+msgstr "ask_user_question 校验失败。"
+
 #: src/iac_code/pipeline/engine/ask_user_question_tool.py
 msgid "ask_user_question requires a pipeline event queue."
 msgstr "ask_user_question 需要 pipeline 事件队列。"
@@ -3166,6 +3170,11 @@ msgstr "conclusion 必须符合以下 schema 摘要:\n"
 msgid "<fill according to the current step requirements>"
 msgstr "<按当前步骤要求填写>"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+#: src/iac_code/ui/renderer.py
+msgid "complete_step validation failed."
+msgstr "complete_step 校验失败。"
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid "Clarification is required before completing the current step."
 msgstr "完成当前步骤前需要先澄清。"

src/iac_code/pipeline/engine/ask_user_question_tool.py

@@ -68,6 +68,15 @@ def is_concurrency_safe(self, tool_input: dict[str, Any]) -> bool:
     def needs_event_queue(self) -> bool:
         return True
 
+    def render_tool_result_message(self, output: str, *, is_error: bool = False, verbose: bool = False) -> str | None:
+        if not output or not is_error:
+            return None
+        if output.startswith("Invalid input for tool 'ask_user_question':"):
+            if verbose:
+                return output.strip()
+            return _("ask_user_question validation failed.")
+        return output.strip()
+
     async def execute(self, *, tool_input: dict[str, Any], context: ToolContext) -> ToolResult:
         if context.event_queue is None:
             return ToolResult.error(_("ask_user_question requires a pipeline event queue."))

src/iac_code/pipeline/engine/complete_step_tool.py

@@ -301,6 +301,13 @@ def _example_from_schema(cls, schema: Any) -> Any:
     def is_read_only(self, input: dict | None = None) -> bool:
         return True
 
+    def render_tool_result_message(self, output: str, *, is_error: bool = False, verbose: bool = False) -> str | None:
+        if not output or not is_error:
+            return None
+        if verbose:
+            return output.strip()
+        return _("complete_step validation failed.")
+
     def _validate_conclusion(self, conclusion: dict) -> str | None:
         """Validate conclusion against schema. Returns error message or None."""
         schema = self._step_config.conclusion_schema

src/iac_code/pipeline/engine/step_executor.py

@@ -415,6 +415,7 @@ def build_agent_loop_context(
         agent_session_id = transcript_id or session_id
         session_usage_store = None
         result_storage_dir = None
+        transcript_image_cache_dir = None
         audit_log_path = None
         if (
             transcript_id
@@ -429,13 +430,22 @@ def build_agent_loop_context(
                     root_session_dir,
                     session_paths.transcript_tool_results_dir(transcript_id),
                 )
+                transcript_image_cache_dir = ensure_session_owned_dir(
+                    root_session_dir,
+                    session_paths.transcript_image_cache_dir(transcript_id),
+                )
                 audit_log_path = session_paths.transcript_permission_audit_path(transcript_id)
                 usage_path = session_paths.transcript_usage_path(transcript_id)
                 ensure_session_owned_parent(root_session_dir, audit_log_path)
                 ensure_session_owned_parent(root_session_dir, usage_path)
                 assert transcript_dir == session_paths.transcript_dir(transcript_id)
                 session_usage_store = SessionUsageStore(path_provider=lambda _cwd, _sid, path=usage_path: path)
         step_skill_roots = self._resolve_step_skill_roots(step)
+        trusted_read_roots = list(step_skill_roots)
+        if result_storage_dir is not None:
+            trusted_read_roots.append(str(result_storage_dir))
+        if transcript_image_cache_dir is not None:
+            trusted_read_roots.append(str(transcript_image_cache_dir))
         agent_loop = AgentLoop(
             provider_manager=self._provider_manager,
             system_prompt=system_prompt,
@@ -453,7 +463,7 @@ def build_agent_loop_context(
             pause_event=self._pause_event,
             permission_context_getter=self._permission_context_getter,
             auto_trigger_skills=self._resolve_auto_trigger_skills(step),
-            tool_context_trusted_read_directories=step_skill_roots,
+            tool_context_trusted_read_directories=trusted_read_roots,
             tool_context_relative_read_directories=step_skill_roots,
             pipeline_mode=True,
             tool_context_env_overrides=self._tool_context_env_overrides,

src/iac_code/services/session_layout.py

@@ -198,3 +198,6 @@ def transcript_permission_audit_path(self, transcript_id: str) -> Path:
 
     def transcript_tool_results_dir(self, transcript_id: str) -> Path:
         return self.transcript_dir(transcript_id) / "tool-results"
+
+    def transcript_image_cache_dir(self, transcript_id: str) -> Path:
+        return self.transcript_dir(transcript_id) / "image-cache"

src/iac_code/tools/cloud/base_api.py

@@ -139,7 +139,15 @@ def render_tool_result_message(self, output: str, *, is_error: bool = False, ver
         if verbose:
             return output.strip()
         action = getattr(self, "_last_action", "")
-        result = getattr(self, "_last_result", None)
+        result = None
+        try:
+            parsed = json.loads(output)
+        except (TypeError, ValueError):
+            parsed = None
+        if isinstance(parsed, dict):
+            result = parsed
+        else:
+            result = getattr(self, "_last_result", None)
         if action and result is not None:
             return self._summarize_success_result(action, result)
         lines = output.strip().splitlines()

src/iac_code/ui/renderer.py

@@ -101,6 +101,20 @@ def _known_tool_result_message(
     is_error: bool = False,
     verbose: bool = False,
 ) -> str | None:
+    if tool_name == "complete_step" and is_error:
+        if verbose:
+            return output.strip()
+        return _("complete_step validation failed.")
+
+    if (
+        tool_name == "ask_user_question"
+        and is_error
+        and output.startswith("Invalid input for tool 'ask_user_question':")
+    ):
+        if verbose:
+            return output.strip()
+        return _("ask_user_question validation failed.")
+
     if tool_name == "ros_deploy":
         from iac_code.pipeline.selling.tools.ros_deploy_tool import RosDeployTool
 
@@ -2090,6 +2104,7 @@ def replay_history(self, messages: list) -> None:
                                 result=result.content if result else None,
                                 is_error=result.is_error if result else False,
                                 done=True,
+                                metadata=copy.deepcopy(result.metadata) if result else None,
                             )
                             segments.append(_Segment(kind="tool", tool=rec))
                 if segments:

tests/agent/test_agent_loop_new.py

@@ -14,6 +14,8 @@
     TOOL_RENDER_DISPLAY_NAME_KEY,
     TOOL_RENDER_METADATA_KEY,
     TOOL_RENDER_RESULT_COMPACT_KEY,
+    TOOL_RENDER_RESULT_VERBOSE_KEY,
+    TOOL_RENDER_VERBOSE_RESULT_IN_TRANSCRIPT_KEY,
     CompactionEvent,
     MessageEndEvent,
     MessageStartEvent,
@@ -101,6 +103,32 @@ def get_stats_snapshot(self):
     assert recall_service.surfaced == {"ros-yaml.md"}
 
 
+def test_tool_result_block_metadata_filters_render_metadata_to_json_safe_fields():
+    sentinel = object()
+
+    metadata = AgentLoop._tool_result_block_metadata(
+        SimpleNamespace(is_externalized=False),
+        {
+            TOOL_RENDER_METADATA_KEY: {
+                TOOL_RENDER_DISPLAY_NAME_KEY: "Custom tool",
+                TOOL_RENDER_RESULT_COMPACT_KEY: sentinel,
+                TOOL_RENDER_RESULT_VERBOSE_KEY: "Verbose output",
+                TOOL_RENDER_VERBOSE_RESULT_IN_TRANSCRIPT_KEY: True,
+                "plugin_payload": sentinel,
+            }
+        },
+    )
+
+    assert metadata == {
+        TOOL_RENDER_METADATA_KEY: {
+            TOOL_RENDER_DISPLAY_NAME_KEY: "Custom tool",
+            TOOL_RENDER_RESULT_VERBOSE_KEY: "Verbose output",
+            TOOL_RENDER_VERBOSE_RESULT_IN_TRANSCRIPT_KEY: True,
+        }
+    }
+    json.dumps(metadata)
+
+
 class TestAgentLoopInit:
     def test_init(self, mock_provider, mock_registry):
         loop = AgentLoop(provider_manager=mock_provider, system_prompt="test", tool_registry=mock_registry)
@@ -1852,6 +1880,12 @@ async def fake_stream(messages, system, tools=None, max_tokens=8192):
                 TOOL_RENDER_RESULT_COMPACT_KEY: "compact result",
             }
         }
+        result_blocks = loop.context_manager.add_tool_results.call_args.args[0]
+        assert result_blocks[0].metadata == {
+            TOOL_RENDER_METADATA_KEY: {
+                TOOL_RENDER_RESULT_COMPACT_KEY: "compact result",
+            }
+        }
         tool_starts = [e for e in events if isinstance(e, ToolUseStartEvent)]
         assert len(tool_starts) == 1
         assert not hasattr(tool_starts[0], "renderer_tool")

tests/pipeline/engine/test_ask_user_question_tool.py

@@ -42,6 +42,19 @@ def test_schema_requires_question_and_options(self):
         assert option_schema["required"] == ["id", "label"]
         assert schema["additionalProperties"] is False
 
+    def test_validation_error_renders_compact_summary(self):
+        long_error = (
+            "Invalid input for tool 'ask_user_question': "
+            "[{'id': 'tech_stack_nodejs', 'label': 'Node.js'}] is not of type 'object'. "
+            "Please provide all required parameters as defined in the tool schema."
+        )
+
+        compact = AskUserQuestionTool().render_tool_result_message(long_error, is_error=True)
+
+        assert compact == "ask_user_question validation failed."
+        assert "tech_stack_nodejs" not in compact
+        assert "not of type" not in compact
+
 
 class TestAskUserQuestionToolExecute:
     @pytest.mark.asyncio