Remove leftover code in the global chat planner

.changeset/planner-shared-stream.md

@@ -0,0 +1,10 @@
+---
+"apollo": patch
+---
+
+Global chat: the planner now streams the whole turn on the stream manager the
+router passes in, instead of replacing it with a second one. That second
+manager emitted its own `message_start` and restarted content block indices
+mid-turn, and repeated the opening spinner. Also removes an unreachable
+`call_job_code_agent` branch left behind when job code calls moved to the
+concurrent path

services/global_chat/planner.py

@@ -101,18 +101,13 @@ def run(
         logger.info("Planner.run() called")
 
         stream_manager = stream_manager or StreamManager(model=self.model, stream=stream)
-        if workflow_yaml:
-            stream_manager.send_thinking(STATUS_REVIEWING_WORKFLOW + STATUS_PLANNING)
-        else:
-            stream_manager.send_thinking(STATUS_NEW_WORKFLOW + STATUS_PLANNING)
 
         self.current_yaml = workflow_yaml
         self.yaml_modified = False
         self._user = user
         self._metrics_opt_in = metrics_opt_in
         self._segments: List[Dict] = []
 
-        stream_manager = StreamManager(model=self.model, stream=stream)
         if workflow_yaml:
             self._send_spinner(stream_manager, STATUS_REVIEWING_WORKFLOW + STATUS_PLANNING)
         else:
@@ -451,64 +446,6 @@ def _execute_tool(self, tool_use_block, stream_manager, total_usage, tool_calls_
 
             tool_calls_meta.append({"tool": "call_workflow_agent", "input": tool_use_block.input})
 
-        elif tool_use_block.name == "call_job_code_agent":
-            job_key = tool_use_block.input.get("job_key")
-
-            # Guard: workflow must exist and contain the target job
-            if not self.current_yaml:
-                tool_result = "ERROR: No workflow exists yet. Call call_workflow_agent first to create the workflow, then call call_job_code_agent."
-                tool_calls_meta.append({"tool": "call_job_code_agent", "input": tool_use_block.input, "skipped": True})
-                return tool_result
-            matched_job_key = None
-            if job_key:
-                matched_job_key, job_data = find_job_in_yaml(self.current_yaml, job_key)
-                if not job_data:
-                    tool_result = f"ERROR: Job key '{job_key}' not found in workflow YAML. Create the workflow with this job first."
-                    tool_calls_meta.append(
-                        {"tool": "call_job_code_agent", "input": tool_use_block.input, "skipped": True}
-                    )
-                    return tool_result
-
-            try:
-                subagent_result = call_job_agent(
-                    tool_use_block.input,
-                    workflow_yaml=self.current_yaml,
-                    api_key=self.api_key,
-                    user=self._user,
-                    metrics_opt_in=self._metrics_opt_in,
-                )
-            except Exception as e:
-                logger.exception("call_job_code_agent failed")
-                tool_calls_meta.append({"tool": "call_job_code_agent", "input": tool_use_block.input, "error": str(e)})
-                return f"ERROR: The job code agent failed: {e}. No code was generated for this job."
-
-            if "usage" in subagent_result:
-                total_usage.update(sum_usage(total_usage, subagent_result["usage"]))
-
-            # Stitch code into live state immediately. Use the YAML key returned by
-            # find_job_in_yaml — `job_key` from the planner may be a fuzzy variant
-            # (case, hyphens vs underscores, or the job's name field), and
-            # stitch_job_code does an exact key match.
-            suggested_code = subagent_result.get("suggested_code")
-            stitched = False
-            if matched_job_key and suggested_code and self.current_yaml:
-                self.current_yaml = stitch_job_code(self.current_yaml, matched_job_key, suggested_code)
-                self.yaml_modified = True
-                stitched = True
-                self._send_yaml(stream_manager)
-                logger.info(f"Stitched code for job '{matched_job_key}' into current_yaml")
-
-            self.subagent_results.append(subagent_result)
-            tool_result = format_subagent_result_for_llm(subagent_result)
-            if stitched:
-                tool_result += "\n\n[Job code generated and stitched into the workflow.]"
-            elif suggested_code:
-                tool_result += "\n\n[Job code was generated but NOT added to the workflow — no job_key matched. Retry with the exact job key.]"
-            else:
-                tool_result += "\n\n[No job code was generated.]"
-
-            tool_calls_meta.append({"tool": "call_job_code_agent", "input": tool_use_block.input})
-
         elif tool_use_block.name == "inspect_job_code":
             # Accept job_keys (list); tolerate legacy single job_key
             job_keys = tool_use_block.input.get("job_keys") or []
@@ -718,13 +655,6 @@ def _tool_status_message(self, tool_use_block) -> str:
                 return "Reviewing the workflow..."
             return "Building workflow outline..."
 
-        if name == "call_job_code_agent":
-            job_key = inputs.get("job_key")
-            display_name = self._display_name_for_job(job_key)
-            if display_name:
-                return f"Writing code for \"{display_name}\"..."
-            return "Writing job code..."
-
         if name == "inspect_job_code":
             job_keys = inputs.get("job_keys") or ([inputs["job_key"]] if inputs.get("job_key") else [])
             display_names = [n for n in (self._display_name_for_job(k) for k in job_keys) if n]

services/global_chat/tests/unit/test_planner.py

@@ -1,5 +1,6 @@
 """Unit tests for PlannerAgent tool execution and user-content building."""
 
+from types import SimpleNamespace
 from unittest.mock import patch
 
 from global_chat.planner import PlannerAgent
@@ -48,13 +49,21 @@ def __init__(self, name: str, tool_input: dict, block_id: str = "tu_1"):
 class StubStreamManager:
     def __init__(self) -> None:
         self.statuses: list[dict] = []
+        self.thinking: list[object] = []
+        self.ended = False
 
-    def send_thinking(self, *_args: object, **_kwargs: object) -> None:
-        pass
+    def send_thinking(self, text: object = None, *_args: object, **_kwargs: object) -> None:
+        self.thinking.append(text)
 
     def send_changes(self, *_args: object, **_kwargs: object) -> None:
         pass
 
+    def send_text(self, *_args: object, **_kwargs: object) -> None:
+        pass
+
+    def end_stream(self, *_args: object, **_kwargs: object) -> None:
+        self.ended = True
+
     def send_status(
         self,
         content: str,
@@ -82,9 +91,9 @@ def test_job_agent_failure_returns_error_tool_result() -> None:
     meta = []
 
     with patch("global_chat.planner.call_job_agent", side_effect=RuntimeError("boom")):
-        result = planner._execute_tool(block, StubStreamManager(), empty_usage(), meta)
+        results = planner._execute_job_code_tools_parallel([block], StubStreamManager(), empty_usage(), meta)
 
-    assert result.startswith("ERROR: The job code agent failed: boom")
+    assert results[0]["content"].startswith("ERROR: The job code agent failed: boom")
     assert meta[0]["error"] == "boom"
 
 
@@ -106,10 +115,10 @@ def test_job_code_without_matched_key_is_reported_as_not_stitched() -> None:
     subagent_result = {"response": "done", "suggested_code": "newCode();", "usage": empty_usage()}
 
     with patch("global_chat.planner.call_job_agent", return_value=subagent_result):
-        result = planner._execute_tool(block, StubStreamManager(), empty_usage(), [])
+        results = planner._execute_job_code_tools_parallel([block], StubStreamManager(), empty_usage(), [])
 
-    assert "NOT added to the workflow" in result
-    assert "stitched into the workflow" not in result
+    assert "NOT added to the workflow" in results[0]["content"]
+    assert "stitched into the workflow" not in results[0]["content"]
     assert planner.current_yaml == WORKFLOW_YAML
     assert planner.yaml_modified is False
 
@@ -200,6 +209,42 @@ def fake_call_job_agent(_tool_input: dict, workflow_yaml: str, *_args: object, *
     assert "newCode();" in planner.current_yaml
 
 
+def test_run_uses_the_stream_manager_the_router_passed() -> None:
+    """The router's manager is the whole turn's stream; run() must not replace it."""
+    planner = make_planner()
+    planner.model = "claude-test"
+    planner.max_tool_calls = 5
+    stream = StubStreamManager()
+
+    response = SimpleNamespace(
+        stop_reason="end_turn",
+        content=[SimpleNamespace(type="text", text="Here you go.")],
+        usage=SimpleNamespace(
+            input_tokens=0,
+            output_tokens=0,
+            cache_creation_input_tokens=0,
+            cache_read_input_tokens=0,
+        ),
+    )
+
+    with patch.object(PlannerAgent, "_build_system_prompt", return_value=[]), \
+         patch.object(PlannerAgent, "_call_api", return_value=response), \
+         patch("global_chat.planner.StreamManager") as stream_manager_cls:
+        result = planner.run(
+            content="add a step",
+            workflow_yaml=WORKFLOW_YAML,
+            page=None,
+            history=[],
+            stream=True,
+            stream_manager=stream,
+        )
+
+    stream_manager_cls.assert_not_called()
+    assert len(stream.thinking) == 1
+    assert stream.ended is True
+    assert result.response == "Here you go."
+
+
 def test_user_content_names_the_step_being_viewed() -> None:
     planner = make_planner()