diff --git a/.changeset/planner-shared-stream.md b/.changeset/planner-shared-stream.md new file mode 100644 index 00000000..86ad332d --- /dev/null +++ b/.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 diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index 677967fe..19493059 100644 --- a/services/global_chat/planner.py +++ b/services/global_chat/planner.py @@ -101,10 +101,6 @@ 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 @@ -112,7 +108,6 @@ def run( 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] diff --git a/services/global_chat/tests/unit/test_planner.py b/services/global_chat/tests/unit/test_planner.py index 6f7dc743..b242360e 100644 --- a/services/global_chat/tests/unit/test_planner.py +++ b/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()