diff --git a/src/orchestrator/control_systems/hybrid_control_system.py b/src/orchestrator/control_systems/hybrid_control_system.py
index 82039535..6d4db4c2 100644
--- a/src/orchestrator/control_systems/hybrid_control_system.py
+++ b/src/orchestrator/control_systems/hybrid_control_system.py
@@ -512,7 +512,14 @@ async def _handle_file_operation(
         if action_text in filesystem_actions and task.parameters:
             # Use UnifiedTemplateResolver for template resolution
             resolved_params = task.parameters.copy()
-            resolved_params["action"] = action_text
+            # Only supply the operation when the step did not state one.
+            # `file` names the TOOL, not an operation, so overwriting a
+            # caller's `action: read` with it sent action="file" to
+            # FileSystemTool and every such step failed with
+            # "Unknown filesystem action: file". The metadata-routed branch
+            # above already guards this; this branch did not.
+            if "action" not in resolved_params:
+                resolved_params["action"] = action_text
             
             # Prepare template context using the unified system
             template_context = self._prepare_template_context(context)
diff --git a/tests/integration/test_pipeline_integration_infrastructure.py b/tests/integration/test_pipeline_integration_infrastructure.py
index 4cac1859..544c2633 100644
--- a/tests/integration/test_pipeline_integration_infrastructure.py
+++ b/tests/integration/test_pipeline_integration_infrastructure.py
@@ -332,12 +332,16 @@ async def test_pipeline_test_provider_get_model(self):
         assert stats['model_usage']["pipeline-test-model"]['requests'] == 1
     
     @pytest.mark.asyncio
-    async def test_pipeline_test_provider_initialization(self):
-        """Test provider initialization process."""
-        
+    async def test_pipeline_test_provider_async_initialization(self):
+        """Test the async initialize() call.
+
+        Renamed: this shared a name with the constructor test above, so it
+        silently replaced it and that test never ran. The two check different
+        things and both are wanted.
+        """
         provider = PipelineTestProvider()
         await provider.initialize()
-        
+
         assert provider.is_initialized is True
     
     def test_pipeline_test_provider_usage_statistics(self):
diff --git a/tests/integration/test_research_assistant_with_report.py b/tests/integration/test_research_assistant_with_report.py
index 0203d581..d5b806d3 100644
--- a/tests/integration/test_research_assistant_with_report.py
+++ b/tests/integration/test_research_assistant_with_report.py
@@ -115,12 +115,14 @@ async def test_report_generator_tool(self):
             recommendations=["Recommendation 1", "Recommendation 2"],
             quality_score=0.85)
 
-        # Verify report
+        # Verify report. Tools return {success, result, error}; the report
+        # itself is in `result` (see #433).
         assert result["success"] is True
-        assert "markdown" in result
-        assert result["word_count"] > 0
+        data = result["result"]
+        assert "markdown" in data
+        assert data["word_count"] > 0
 
-        markdown = result["markdown"]
+        markdown = data["markdown"]
         assert "# Test Research Report" in markdown
         assert "Finding 1" in markdown
         assert "Finding 2" in markdown
@@ -170,10 +172,12 @@ async def test_pdf_compiler_tool(self):
             # Verify result
             if result["success"]:
                 assert output_path.exists()
-                assert result["file_size"] > 0
+                assert result["result"]["file_size"] > 0
             else:
-                # PDF generation failed (likely pandoc not available)
-                assert "error" in result
+                # PDF generation failed (likely pandoc not available).
+                # `assert "error" in result` could never fail -- the envelope
+                # always carries that key. Assert the value instead.
+                assert result["error"], "a failed compile must say why"
 
     @pytest.mark.integration
     @pytest.mark.asyncio
@@ -194,12 +198,23 @@ async def test_web_search_integration(self):
         # Perform real search
         result = await tool.execute(query="Python asyncio tutorial", max_results=5)
 
-        # Verify results
-        assert "results" in result
-        assert len(result["results"]) > 0
+        # Tools return {success, result, error}; the payload is in `result`.
+        assert result["success"] is True, f"search failed: {result['error']}"
+        data = result["result"]
+        assert "results" in data
+
+        if not data["results"]:
+            # No backend reachable (missing [web] extra, or no network). This
+            # test is about the shape of a result, so an unavailable search
+            # engine is a missing prerequisite rather than a defect.
+            pytest.skip(
+                f"web search returned nothing "
+                f"(total_results={data.get('total_results')}); "
+                f"no search backend available"
+            )
 
         # Check result structure
-        first_result = result["results"][0]
+        first_result = data["results"][0]
         assert "snippet" in first_result
         assert "rank" in first_result
         assert "relevance" in first_result
diff --git a/tests/integration/test_tools_real_world.py b/tests/integration/test_tools_real_world.py
index 095ff39e..8a661464 100644
--- a/tests/integration/test_tools_real_world.py
+++ b/tests/integration/test_tools_real_world.py
@@ -26,6 +26,40 @@
 pytestmark = pytest.mark.integration
 
 
+def payload(response):
+    """Return the data half of a tool's response.
+
+    Every tool returns the envelope ``{"success": bool, "result": Any,
+    "error": str | None}``. ``success`` and ``error`` are read from the
+    envelope directly; everything else -- ``stdout``, ``content``, ``items``,
+    ``markdown`` and so on -- lives one level down, in ``result``.
+
+    These tests originally asserted against a flat shape that predates the
+    envelope, which is why they failed with ``KeyError`` on keys that did
+    exist (see #433).
+    """
+    assert isinstance(response, dict), (
+        f"tool returned {type(response).__name__}, expected the response envelope"
+    )
+    assert "result" in response, (
+        f"no 'result' key in tool response; got {sorted(response)}"
+    )
+    return response["result"]
+
+
+def skip_if_optional_dependency_missing(response):
+    """Skip when a tool could not run for lack of an optional package.
+
+    A missing ``[web]`` extra is a statement about this environment, not a
+    defect in the tool, so it must skip with a reason rather than fail --
+    the same rule the rest of the suite follows.
+    """
+    error = str(response.get("error") or "")
+    lowered = error.lower()
+    if "no module named" in lowered or "playwright is required" in lowered:
+        pytest.skip(f"optional dependency missing: {error}")
+
+
 @pytest.fixture(scope="module")
 def model_registry():
     """Initialize real models for testing."""
@@ -65,37 +99,38 @@ async def test_scrape_real_website(self, browser_tool):
         # Use example.com - it's stable and simple
         result = await browser_tool.execute(url="https://example.com", action="scrape")
 
-        # Check if there was an error
-        if "error" in result:
+        skip_if_optional_dependency_missing(result)
+        # `"error" in result` is always true -- the envelope always carries the
+        # key, with None on success. The error's *value* is the signal.
+        if result["error"]:
             pytest.fail(f"Scraping failed: {result['error']}")
 
         # Should have scraped content
-        assert "url" in result
-        assert result["url"] == "https://example.com"
-        assert "title" in result
-        assert "Example Domain" in result["title"]
+        data = payload(result)
+        assert data["url"] == "https://example.com"
+        assert "Example Domain" in data["title"]
 
         # Check for text content (if extracted)
-        if "text" in result:
-            assert "Example Domain" in result["text"]
-            assert "More information" in result["text"]
+        if "text" in data:
+            assert "Example Domain" in data["text"]
+            assert "More information" in data["text"]
 
         # Check for other metadata
-        if "status_code" in result:
-            assert result["status_code"] == 200
+        if "status_code" in data:
+            assert data["status_code"] == 200
 
     @pytest.mark.asyncio
     async def test_verify_real_website(self, browser_tool):
         """Test verifying a real website."""
         result = await browser_tool.execute(url="https://example.com", action="verify")
 
-        # Check if there was an error
-        if "error" in result:
+        skip_if_optional_dependency_missing(result)
+        if result["error"]:
             pytest.fail(f"Verification failed: {result['error']}")
 
-        assert "url" in result
-        assert "accessible" in result
-        assert result["accessible"] is True
+        data = payload(result)
+        assert "url" in data
+        assert data["accessible"] is True
 
     @pytest.mark.asyncio
     async def test_invalid_url_handling(self, browser_tool):
@@ -104,9 +139,15 @@ async def test_invalid_url_handling(self, browser_tool):
             url="https://this-domain-definitely-does-not-exist-12345.com",
             action="scrape")
 
-        # Should have an error for invalid URL
-        assert "error" in result
-        assert "url" in result
+        # Skip first: a missing bs4 also produces an error, which would make
+        # this pass while proving nothing about URL handling.
+        skip_if_optional_dependency_missing(result)
+
+        # `assert "error" in result` used to stand here. The envelope always
+        # has that key, so the assertion could never fail -- it passed whether
+        # or not the tool handled the bad URL at all. Assert the value.
+        assert result["success"] is False
+        assert result["error"], "an unreachable domain must report an error"
 
     @pytest.mark.asyncio
     @pytest.mark.timeout(240)  # Increased timeout to allow for playwright installation
@@ -122,7 +163,8 @@ async def test_scrape_with_javascript(self, browser_tool):
         elapsed = time.time() - start_time
         
         # The tool should handle playwright installation automatically
-        if "error" in result:
+        skip_if_optional_dependency_missing(result)
+        if result["error"]:
             # Get more diagnostic info
             error_msg = result["error"]
             
@@ -153,8 +195,9 @@ async def test_scrape_with_javascript(self, browser_tool):
                 pytest.fail(f"JS scraping failed after {elapsed:.1f}s: {error_msg}")
 
         # Should have scraped content
-        assert "url" in result
-        assert "title" in result
+        data = payload(result)
+        assert "url" in data
+        assert "title" in data
 
 
 class TestWebSearchTool:
@@ -172,22 +215,22 @@ async def test_real_web_search(self, search_tool):
             query="Python programming language official documentation", max_results=5
         )
 
-        # WebSearchTool returns different format - no "success" key
-        assert "results" in result
-        assert "query" in result
-        assert result["query"] == "Python programming language official documentation"
+        data = payload(result)
+        assert "results" in data
+        assert data["query"] == "Python programming language official documentation"
 
         # Check if we got results (may be empty if search fails)
-        if result.get("error"):
-            # Search failed, but should still have proper structure
-            assert result["total_results"] == 0
+        if result["error"] or not data["results"]:
+            # Search failed or returned nothing -- it must still be
+            # structurally coherent rather than claiming results it lacks.
+            assert data["total_results"] == 0
         else:
             # Should have some results
-            assert len(result["results"]) > 0
-            assert len(result["results"]) <= 5
+            assert len(data["results"]) > 0
+            assert len(data["results"]) <= 5
 
             # Check result structure
-            for item in result["results"]:
+            for item in data["results"]:
                 assert "title" in item
                 assert "url" in item
                 assert "snippet" in item
@@ -199,9 +242,15 @@ async def test_empty_query_handling(self, search_tool):
         """Test handling of empty search query."""
         result = await search_tool.execute(query="", max_results=5)
 
-        # Should have error for empty query
-        assert "error" in result
-        assert result["total_results"] == 0
+        # The tool rejects the query outright rather than inventing an empty
+        # result set, so there is no payload to inspect -- `result` is None.
+        # The old `assert "error" in result` could never fail, because the
+        # envelope always carries that key.
+        assert result["success"] is False
+        assert "no query" in str(result["error"]).lower()
+        assert result["result"] is None, (
+            "a rejected query must not come back with a fabricated payload"
+        )
 
 
 class TestTerminalTool:
@@ -218,9 +267,11 @@ async def test_simple_command_execution(self, terminal_tool):
         # Test echo command
         result = await terminal_tool.execute(command="echo 'Hello from terminal tool'")
 
+        # `success` is the envelope's own key and tracks the command's exit
+        # status; the stream contents live in the payload.
         assert result["success"] is True
-        assert result["stdout"].strip() == "Hello from terminal tool"
-        assert result["return_code"] == 0
+        assert payload(result)["stdout"].strip() == "Hello from terminal tool"
+        assert payload(result)["return_code"] == 0
 
     @pytest.mark.asyncio
     async def test_command_with_error(self, terminal_tool):
@@ -228,8 +279,8 @@ async def test_command_with_error(self, terminal_tool):
         result = await terminal_tool.execute(command="ls /nonexistent/directory/path")
 
         assert result["success"] is False
-        assert result["return_code"] != 0
-        assert result["stderr"] != ""  # Should have error message
+        assert payload(result)["return_code"] != 0
+        assert payload(result)["stderr"] != ""  # Should have error message
 
     @pytest.mark.asyncio
     async def test_command_timeout(self, terminal_tool):
@@ -251,7 +302,7 @@ async def test_working_directory(self, terminal_tool, temp_workspace):
         )
 
         assert result["success"] is True
-        assert "test.txt" in result["stdout"]
+        assert "test.txt" in payload(result)["stdout"]
 
 
 class TestFileSystemTool:
@@ -280,7 +331,7 @@ async def test_file_operations(self, fs_tool, temp_workspace):
         read_result = await fs_tool.execute(action="read", path=str(test_file))
 
         assert read_result["success"] is True
-        assert read_result["content"] == test_content
+        assert payload(read_result)["content"] == test_content
 
         # Test delete
         delete_result = await fs_tool.execute(action="delete", path=str(test_file))
@@ -311,8 +362,9 @@ async def test_directory_operations(self, fs_tool, temp_workspace):
         list_result = await fs_tool.execute(action="list", path=str(test_dir))
 
         assert list_result["success"] is True
-        assert len(list_result["items"]) == 3  # dummy.txt, file1.txt, file2.txt
-        file_names = [item["name"] for item in list_result["items"]]
+        items = payload(list_result)["items"]
+        assert len(items) == 3  # dummy.txt, file1.txt, file2.txt
+        file_names = [item["name"] for item in items]
         assert "file1.txt" in file_names
         assert "file2.txt" in file_names
         assert "dummy.txt" in file_names
@@ -323,7 +375,8 @@ async def test_file_not_found(self, fs_tool):
         result = await fs_tool.execute(action="read", path="/nonexistent/file/path.txt")
 
         assert result["success"] is False
-        assert "error" in result
+        # Was `assert "error" in result`, which the envelope makes always true.
+        assert result["error"], "a missing file must report why it failed"
 
 
 class TestDataProcessingTool:
@@ -356,11 +409,13 @@ async def test_json_processing(self, data_tool, temp_workspace):
         )
 
         assert result["success"] is True
-        assert "result" in result
-        assert result["original_count"] == 3
-        assert result["filtered_count"] == 1
-        assert len(result["result"]) == 1
-        assert result["result"][0]["name"] == "Alice"
+        data = payload(result)
+        assert data["original_count"] == 3
+        assert data["filtered_count"] == 1
+        # The filtered rows are under `data`, not directly under the envelope's
+        # `result` -- the old assertion happened to read the envelope key.
+        assert len(data["data"]) == 1
+        assert data["data"][0]["name"] == "Alice"
 
     @pytest.mark.asyncio
     async def test_csv_processing(self, data_tool, temp_workspace):
@@ -380,11 +435,11 @@ async def test_csv_processing(self, data_tool, temp_workspace):
         )
 
         assert result["success"] is True
-        assert "result" in result
-        assert result["target_format"] == "json"
+        data = payload(result)
+        assert data["target_format"] == "json"
 
         # Parse the JSON result
-        converted_data = json.loads(result["result"])
+        converted_data = json.loads(data["data"])
         assert len(converted_data) == 3
         assert converted_data[0]["name"] == "Alice"
         assert converted_data[1]["age"] == "25"  # Note: CSV values are strings
@@ -492,10 +547,11 @@ async def test_markdown_report_generation(self, report_tool, temp_workspace):
             output_path=str(temp_workspace / "report.md"))
 
         assert result["success"] is True
-        assert "markdown" in result
+        data = payload(result)
+        assert "markdown" in data
 
         # Verify content structure
-        content = result["markdown"]
+        content = data["markdown"]
         assert "# AI Applications Report" in content
         assert "## Executive Summary" in content
         assert "## Search Results" in content
@@ -543,10 +599,10 @@ async def test_pdf_generation(self, pdf_tool, temp_workspace):
             # Pandoc is installed
             assert output_path.exists()
             assert output_path.stat().st_size > 0
-            assert result["file_size"] > 0
+            assert payload(result)["file_size"] > 0
         else:
             # Pandoc not installed - should fail gracefully
-            assert "pandoc" in result["error"].lower()
+            assert "pandoc" in str(result["error"]).lower()
 
 
 class TestToolIntegration:
@@ -619,6 +675,12 @@ async def test_web_scraping_pipeline(self, orchestrator, temp_workspace):
   search_count: "{{search_web.total_results}}"
 """
 
+        # This pipeline needs a real web-search backend AND a real model for
+        # the `generate` step. Without the [web] extra the search step fails
+        # and the whole pipeline aborts, which says nothing about the
+        # orchestration being tested.
+        pytest.importorskip("bs4", reason="web scraping needs the [web] extra")
+
         # Execute pipeline
         context = {
             "topic": "artificial intelligence",
@@ -627,7 +689,7 @@ async def test_web_scraping_pipeline(self, orchestrator, temp_workspace):
 
         import time
         start_time = time.time()
-        
+
         try:
             result = await orchestrator.execute_yaml(
                 yaml_content=yaml_content, context=context
@@ -738,7 +800,11 @@ async def test_file_processing_pipeline(self, orchestrator, temp_workspace):
     action: process
     parameters:
       action: transform
-      data: "{{read_data.content}}"
+      # A step's output is the {success, result, error} envelope, so the
+      # payload is reached through `.result`. `{{read_data.content}}` silently
+      # rendered as the literal placeholder text, which then reached
+      # json.loads() and failed there instead -- see #153.
+      data: "{{read_data.result.content}}"
       transform_spec:
         total_price: "sum(item['price'] for item in json.loads(data)['items'])"
         item_count: "len(json.loads(data)['items'])"
@@ -748,7 +814,10 @@ async def test_file_processing_pipeline(self, orchestrator, temp_workspace):
   - id: validate_results
     action: validate
     parameters:
-      data: "{{process_data.processed_data}}"
+      # `| tojson` is required: without it a dict is interpolated with
+      # Python's repr (single quotes), which the validator rejects as
+      # "not JSON or CSV".
+      data: "{{ process_data.processed_data | tojson }}"
       schema:
         type: object
         properties:
@@ -766,11 +835,11 @@ async def test_file_processing_pipeline(self, orchestrator, temp_workspace):
     parameters:
       action: write
       path: "{{output_file}}"
-      content: "{{process_data.processed_data}}"
+      content: "{{ process_data.processed_data | tojson }}"
     depends_on: [validate_results]
 
 outputs:
-  validation_passed: "{{validate_results.valid}}"
+  validation_passed: "{{validate_results.result.valid}}"
   total_price: "{{process_data.processed_data.total_price}}"
 """
 
@@ -786,12 +855,20 @@ async def test_file_processing_pipeline(self, orchestrator, temp_workspace):
 
         # Verify results
         # Check if result has the new structure with steps/outputs
-        if "steps" in result:
-            assert result["steps"]["validate_results"]["valid"] is True
-        else:
-            assert result["validate_results"]["valid"] is True
+        step = (
+            result["steps"]["validate_results"]
+            if "steps" in result
+            else result["validate_results"]
+        )
+        assert step["success"] is True, f"validation step failed: {step['error']}"
+        assert payload(step)["valid"] is True
         assert (temp_workspace / "output_data.json").exists()
 
+        # The saved file must be real JSON, not a Python repr.
+        saved = json.loads((temp_workspace / "output_data.json").read_text())
+        assert saved["item_count"] == 3
+        assert saved["total_price"] == pytest.approx(51.25)
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v", "-s"])
diff --git a/tests/test_filesystem_step_action.py b/tests/test_filesystem_step_action.py
new file mode 100644
index 00000000..c60a1739
--- /dev/null
+++ b/tests/test_filesystem_step_action.py
@@ -0,0 +1,102 @@
+"""A pipeline step must keep the tool operation it asked for.
+
+A step carries two different `action` values that are easy to confuse:
+
+- the **step** action (`action: file`) selects which tool runs;
+- the **parameter** action (`parameters.action: read`) selects what that tool
+  does.
+
+The control system used to copy the first over the second, so `FileSystemTool`
+received `action="file"` -- not one of its operations -- and every such step
+failed with "Unknown filesystem action: file". Reading a file from a pipeline
+was impossible.
+
+Real files, real tool, no mocks.
+"""
+
+import json
+
+import pytest
+
+from orchestrator.control_systems.hybrid_control_system import HybridControlSystem
+from orchestrator.core.task import Task
+from orchestrator.models.model_registry import ModelRegistry
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture
+def control_system():
+    """A control system with an empty registry -- no model is needed here."""
+    return HybridControlSystem(model_registry=ModelRegistry())
+
+
+def _task(step_action, **parameters):
+    """Build a task. `step_action` is the routing action, kwargs the params.
+
+    Named `step_action` rather than `action` precisely because the two
+    collide -- which is the confusion this whole module exists to pin down.
+    """
+    return Task(id="step", name="step", action=step_action, parameters=parameters)
+
+
+@pytest.mark.asyncio
+async def test_read_operation_survives_a_file_step_action(control_system, tmp_path):
+    """The regression: `action: file` must not overwrite `action: read`."""
+    target = tmp_path / "data.json"
+    target.write_text(json.dumps({"items": [1, 2, 3]}))
+
+    result = await control_system._handle_file_operation(
+        _task("file", action="read", path=str(target)), {}
+    )
+
+    assert result["success"] is True, f"read failed: {result['error']}"
+    assert result["result"]["action"] == "read", (
+        "the step's routing action overwrote the requested operation"
+    )
+    assert json.loads(result["result"]["content"]) == {"items": [1, 2, 3]}
+
+
+@pytest.mark.asyncio
+async def test_write_operation_survives_a_file_step_action(control_system, tmp_path):
+    target = tmp_path / "out.txt"
+
+    result = await control_system._handle_file_operation(
+        _task("file", action="write", path=str(target), content="written"), {}
+    )
+
+    assert result["success"] is True, f"write failed: {result['error']}"
+    assert target.read_text() == "written"
+
+
+@pytest.mark.asyncio
+async def test_list_operation_survives_a_file_step_action(control_system, tmp_path):
+    (tmp_path / "a.txt").write_text("a")
+    (tmp_path / "b.txt").write_text("b")
+
+    result = await control_system._handle_file_operation(
+        _task("file", action="list", path=str(tmp_path)), {}
+    )
+
+    assert result["success"] is True, f"list failed: {result['error']}"
+    assert {item["name"] for item in result["result"]["items"]} == {"a.txt", "b.txt"}
+
+
+@pytest.mark.asyncio
+async def test_step_action_still_supplies_the_operation_when_absent(
+    control_system, tmp_path
+):
+    """The fix must not break the case it was originally written for.
+
+    With no `parameters.action`, the step action IS the operation, and that
+    behaviour has to survive.
+    """
+    target = tmp_path / "data.txt"
+    target.write_text("hello")
+
+    result = await control_system._handle_file_operation(
+        _task("read", path=str(target)), {}
+    )
+
+    assert result["success"] is True, f"read failed: {result['error']}"
+    assert result["result"]["content"] == "hello"