From f1df29cf862e90f633bd3905681c52af38eaeba6 Mon Sep 17 00:00:00 2001 From: Hanna Paasivirta Date: Wed, 29 Jul 2026 11:17:58 +0100 Subject: [PATCH 1/5] Global chat: Add answer streaming to planner (#573) * add segments * adjust events * add event type * adjust streaming * update readme * changeset --------- Co-authored-by: Joe Clark --- .changeset/short-months-look.md | 6 + services/global_chat/PAYLOAD_SPEC.md | 31 +++- services/global_chat/README.md | 24 ++- services/global_chat/global_chat.py | 1 + services/global_chat/planner.py | 170 +++++++++++++----- services/global_chat/router.py | 4 + .../global_chat/tests/unit/test_planner.py | 19 +- services/streaming_util.py | 18 ++ 8 files changed, 215 insertions(+), 58 deletions(-) create mode 100644 .changeset/short-months-look.md diff --git a/.changeset/short-months-look.md b/.changeset/short-months-look.md new file mode 100644 index 00000000..2dcff69f --- /dev/null +++ b/.changeset/short-months-look.md @@ -0,0 +1,6 @@ +--- +"apollo": minor +--- + +global chat: add answer streaming to the planner, breaking up responses into +chunks which can be rendered earlier diff --git a/services/global_chat/PAYLOAD_SPEC.md b/services/global_chat/PAYLOAD_SPEC.md index b8ec9846..f8cde2e0 100644 --- a/services/global_chat/PAYLOAD_SPEC.md +++ b/services/global_chat/PAYLOAD_SPEC.md @@ -84,7 +84,13 @@ This document defines the input and output payload structure for the Global Agen ```json { - "response": "string", // Main text response + "response": "string", // Main text response (final answer) + + "response_segments": [ // Durable transcript of the turn, in stream order + { "type": "text", "content": "I'll add the step..." }, + { "type": "status", "content": "Edited workflow structure" }, + { "type": "text", "content": "Done! I added..." } + ], "attachments": [ // Artifacts produced this turn { @@ -96,8 +102,8 @@ This document defines the input and output payload structure for the Global Agen "history": [ // Conversation history including this turn { "role": "user|assistant", - "content": "string | array" // string for direct routes; array of content - } // blocks (text, tool_use, tool_result) for planner path + "content": "string" + } ], "usage": { // Token usage (aggregated across all agents) @@ -125,11 +131,26 @@ This document defines the input and output payload structure for the Global Agen ### Field Descriptions -- **`response`** (string): The main text response from the agent. +- **`response`** (string): The main text response from the agent — the final answer. On the planner path this is the text of the planner's last round only (narration from earlier rounds is not included). + +- **`response_segments`** (array): The durable transcript of the turn in the order it was streamed, as `{"type", "content"}` objects. Two segment types: + - `text` — a text block from the model. On the planner path there is one per round of the tool-calling loop; the last one equals `response`. + - `status` — a completed-action status line ("Edited workflow structure", "Wrote code for \"Fetch Patients\""). Only these settled lines are recorded; the transient "...ing" spinners shown while an action runs are never persisted. + + This lets the frontend persist and re-render the woven view after a page reload without reconstructing it from stream events. On direct routes (workflow_agent, job_code_agent) it is a single `text` segment wrapping `response` — statuses emitted internally by those subagents are not captured. + +#### Streaming status events (planner path) + +Apollo classifies status messages by event type so the client never has to infer their meaning from the text: + +- **`thinking` events** (standard Anthropic thinking blocks) — transient progress spinners ("Reviewing the workflow...", "Writing code for \"X\"..."). Render live; each new status replaces the previous one; drop when the next text block starts. Never persist these. +- **`status` events** (custom event, like `changes`) — completed-action lines. Payload is `{"type": "status", "content": "Edited workflow structure"}`, identical to a `response_segments` entry, so live events and reloaded segments render through the same code path. Persist these (they resolve the preceding spinner). + +Each tool beat streams as: `thinking` spinner → `changes` (if the workflow was modified) → `status` settled line → narration text. - **`attachments`** (array): Artifacts produced during this turn. Each entry has a `type` and `content` field. An empty list `[]` means no artifacts were produced (e.g. a purely informational response). The only supported type is `workflow_yaml`: the full workflow YAML with any job code changes stitched in. Job code edits are never returned separately — the YAML is the single source of truth, which allows multi-step changes in one response. -- **`history`** (array): Updated conversation history including the latest exchange. On direct routes (workflow_agent, job_code_agent), each entry has `content` as a string. On the planner path, entries may have `content` as an array of content blocks (`text`, `tool_use`, `tool_result`) — this is the raw Anthropic messages format from the tool-calling loop. +- **`history`** (array): Updated conversation history including the latest exchange. Each entry has `content` as a string on every route. On the planner path the assistant entry contains only the final answer text — the pre-tool narration segments in `response` are not persisted to history. - **`usage`** (object): Token usage aggregated across all agents invoked (router + planner + sub-agents). diff --git a/services/global_chat/README.md b/services/global_chat/README.md index 3531303c..34b7d4e0 100644 --- a/services/global_chat/README.md +++ b/services/global_chat/README.md @@ -70,6 +70,12 @@ Request to build a new multi-step workflow from scratch: ```json { "response": "I've created a workflow that fetches patient data from CommCare and loads it to DHIS2. The first job retrieves patient records via the CommCare REST API, and the second job maps and uploads them to DHIS2.", + "response_segments": [ + { "type": "text", "content": "I'll build the workflow structure first." }, + { "type": "status", "content": "Built workflow outline" }, + { "type": "status", "content": "Wrote code for \"Fetch from CommCare\", \"Load to DHIS2\"" }, + { "type": "text", "content": "I've created a workflow that fetches patient data from CommCare and loads it to DHIS2..." } + ], "attachments": [ { "type": "workflow_yaml", @@ -106,12 +112,18 @@ Request to build a new multi-step workflow from scratch: **Response Fields:** -- `response`: The assistant's text response to the user +- `response`: The assistant's text response to the user — the final answer. On + the planner path this is the last round's text only; earlier narration lives + in `response_segments` +- `response_segments`: The durable transcript of the turn in stream order — + `text` segments woven with settled `status` lines ("Edited workflow + structure"). Transient "...ing" spinners are not included. See + `PAYLOAD_SPEC.md` for the full streaming event contract - `attachments`: Artifacts produced — currently always `{type: "workflow_yaml", content: string}` when YAML was generated or modified -- `history`: Updated conversation history including the latest exchange. Direct - routes return string `content`; the planner path may return `content` as - Anthropic content-block arrays (`tool_use`/`tool_result`) +- `history`: Updated conversation history including the latest exchange. + `content` is a string on every route; on the planner path the assistant entry + contains only the final answer text - `usage`: Aggregated token usage across all agents called during the request - `meta.agents`: Ordered list of agents invoked (e.g. `["router", "workflow_agent"]` or @@ -151,7 +163,7 @@ For straightforward requests, the router calls subagents directly: ### Planner -For complex requests, the `PlannerAgent` (Claude Sonnet) runs an agentic +For complex requests, the `PlannerAgent` (Claude Opus) runs an agentic tool-calling loop with access to four tools: - **`call_workflow_agent`** — create or modify workflow YAML structure @@ -165,7 +177,7 @@ then calls `call_job_code_agent` for each job that needs code. Job code is stitched into the workflow YAML immediately after each call. The loop continues until the model signals it is done (up to a configurable -maximum of tool calls, default 25). +maximum of tool calls, currently 10 in `config.yaml`). ## Testing diff --git a/services/global_chat/global_chat.py b/services/global_chat/global_chat.py index 57ffec45..e849809c 100644 --- a/services/global_chat/global_chat.py +++ b/services/global_chat/global_chat.py @@ -122,6 +122,7 @@ def main(data_dict: dict) -> dict: # 5. Return structured response return { "response": result.response, + "response_segments": result.response_segments, "attachments": result.attachments, "history": result.history, "usage": result.usage, diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index e69da752..5fec69ff 100644 --- a/services/global_chat/planner.py +++ b/services/global_chat/planner.py @@ -38,6 +38,7 @@ class PlannerResult: """Result from planner run.""" response: str + response_segments: List[Dict] attachments: List[Dict] history: List[Dict] usage: Dict @@ -66,6 +67,7 @@ def __init__(self, config_loader: ConfigLoader, api_key: Optional[str] = None): self.current_yaml: Optional[str] = None self.subagent_results = [] + self._segments: List[Dict] = [] logger.info(f"PlannerAgent initialized with model: {self.model}") @@ -95,16 +97,17 @@ def run( """ logger.info("Planner.run() called") - stream_manager = 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: + self._send_spinner(stream_manager, STATUS_NEW_WORKFLOW + STATUS_PLANNING) system_prompt = self._build_system_prompt() @@ -121,12 +124,10 @@ def run( "cache_read_input_tokens": 0, } - final_text = "" - try: while tool_call_count < self.max_tool_calls: try: - response, buffered_text = self._call_api(system_prompt, messages, stream) + response = self._call_api(system_prompt, messages, stream, stream_manager) for field in [ "input_tokens", @@ -138,17 +139,14 @@ def run( logger.info(f"Claude API call {tool_call_count + 1}: stop_reason={response.stop_reason}") - if response.stop_reason == "end_turn": - # Send final YAML before text, matching workflow_chat/job_chat pattern - if self.yaml_modified and self.current_yaml: - stream_manager.send_changes({"yaml": self.current_yaml}) - - # Flush buffered text chunks - for chunk in buffered_text: - stream_manager.send_text(chunk) + # Text from every round is part of the answer the user saw + # (tool rounds may narrate before calling tools). + round_text = self._extract_text(response) + if round_text: + self._segments.append({"type": "text", "content": round_text}) - final_text = self._extract_text(response) - messages.append({"role": "assistant", "content": final_text}) + if response.stop_reason == "end_turn": + messages.append({"role": "assistant", "content": round_text}) logger.info(f"Tool loop completed. Total calls: {tool_call_count}") break @@ -196,11 +194,20 @@ def run( raise ApolloError(500, f"Tool execution error: {str(e)}") if response.stop_reason != "end_turn": - final_text = self._extract_text(response) logger.warning(f"Loop exited without end_turn (reason: {response.stop_reason})") finally: stream_manager.end_stream() + # The full transcript in stream order: text segments (one per round) + # interleaved with the status messages shown between them, so the + # client can persist and re-render the woven view. + response_segments = self._segments + + # response and history keep only the last round's text (the actual + # answer), matching the direct routes and what was saved before + # narration was streamed. The narration survives in response_segments. + final_text = round_text + if not final_text: stop_reason = getattr(response, "stop_reason", None) if tool_call_count >= self.max_tool_calls: @@ -241,6 +248,7 @@ def run( return PlannerResult( response=final_text, + response_segments=response_segments, attachments=attachments, history=return_history, usage=total_usage, @@ -274,8 +282,14 @@ def _build_user_content(self, content: str, page: Optional[str]) -> str: return user_content - def _call_api(self, system_prompt, messages, stream): - """Make Claude API call. When streaming, buffers text deltas for the caller to flush. + def _call_api(self, system_prompt, messages, stream, stream_manager): + """Make Claude API call. When streaming, forwards text deltas live. + + All text blocks stream to the client as they generate — including the + narration the model writes before tool calls. Each round's text lands + in its own content block (the status and changes events sent between + rounds close the open text block), so the client can weave text and + status events with its own formatting. Adaptive thinking is enabled for better reasoning but thinking content is not streamed to the client — it exposes internal details like tool @@ -283,8 +297,6 @@ def _call_api(self, system_prompt, messages, stream): task-specific status messages sent before each tool execution. """ if stream: - buffered_text = [] - with self.client.messages.stream( model=self.model, max_tokens=self.max_tokens, @@ -295,10 +307,9 @@ def _call_api(self, system_prompt, messages, stream): output_config={"effort": "medium"}, ) as stream_obj: for event in stream_obj: - if event.type == "content_block_delta": - if event.delta.type == "text_delta": - buffered_text.append(event.delta.text) - return stream_obj.get_final_message(), buffered_text + if event.type == "content_block_delta" and event.delta.type == "text_delta": + stream_manager.send_text(event.delta.text) + return stream_obj.get_final_message() else: response = self.client.beta.messages.create( model=self.model, @@ -325,13 +336,48 @@ def _call_api(self, system_prompt, messages, stream): ] }, ) - return response, [] + return response + + def _send_yaml(self, stream_manager) -> None: + """Stream the current YAML as a changes event. + + Called wherever the YAML is actually updated (workflow edit, job-code + stitch), so each change reaches the client the moment it happens — e.g. + a newly added step renders before its code is written. No-op in + non-streaming mode, where the final payload's attachment carries the + YAML instead. + """ + stream_manager.send_changes({"yaml": self.current_yaml}) + + def _send_spinner(self, stream_manager, status: str | list[str]) -> None: + """Send a transient "...ing" spinner as a thinking event. + + Thinking events are live progress only: the client replaces each one + with the next status and never persists them, so spinners are not + recorded in the transcript. + """ + stream_manager.send_thinking(status) + + def _send_settled(self, stream_manager, content: str | None) -> None: + """Send a completed-action line ("Edited workflow structure") as a + custom `status` event and record it in the transcript. + + Unlike spinners, these are durable facts about what happened: the + client persists them (they resolve the preceding spinner), and they + are recorded in `response_segments` so a page reload re-renders the + same view. None means the action left nothing worth showing (e.g. a + consult that changed nothing) — nothing is sent or recorded. + """ + if not content: + return + stream_manager.send_status(content) + self._segments.append({"type": "status", "content": content}) def _find_all_tool_uses(self, content): """Find all tool_use blocks in response content.""" return [block for block in content if block.type == "tool_use"] - def _execute_tool(self, tool_use_block, total_usage, tool_calls_meta) -> str: + def _execute_tool(self, tool_use_block, stream_manager, total_usage, tool_calls_meta) -> str: """Execute a single tool call and return the result string.""" if tool_use_block.name == "search_documentation": tool_result = search_documentation_tool(tool_use_block.input) @@ -355,10 +401,11 @@ def _execute_tool(self, tool_use_block, total_usage, tool_calls_meta) -> str: if "usage" in subagent_result: total_usage.update(sum_usage(total_usage, subagent_result["usage"])) - # Update live state eagerly + # Update live state and stream the change in the same breath if subagent_result.get("response_yaml"): self.current_yaml = subagent_result["response_yaml"] self.yaml_modified = True + self._send_yaml(stream_manager) self.subagent_results.append(subagent_result) @@ -420,6 +467,7 @@ def _execute_tool(self, tool_use_block, total_usage, tool_calls_meta) -> str: 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) @@ -478,8 +526,10 @@ def _execute_tool_blocks(self, tool_use_blocks, stream_manager, total_usage, too tool_results = [] for tool_use_block in other_blocks: - stream_manager.send_thinking(self._tool_status_message(tool_use_block)) - tool_result = self._execute_tool(tool_use_block, total_usage, tool_calls_meta) + self._send_spinner(stream_manager, self._tool_status_message(tool_use_block)) + yaml_before = self.current_yaml + tool_result = self._execute_tool(tool_use_block, stream_manager, total_usage, tool_calls_meta) + self._send_settled(stream_manager, self._settled_status_message(tool_use_block, yaml_before)) tool_results.append( {"type": "tool_result", "tool_use_id": tool_use_block.id, "content": tool_result} ) @@ -506,7 +556,7 @@ def _execute_job_code_tools_parallel(self, blocks, stream_manager, total_usage, status = f"Writing code for {joined}..." else: status = "Writing job code..." - stream_manager.send_thinking(status) + self._send_spinner(stream_manager, status) # Validate and prepare — skip invalid ones before launching threads. # matched_keys carries the YAML key resolved by find_job_in_yaml's @@ -568,6 +618,7 @@ def _execute_job_code_tools_parallel(self, blocks, stream_manager, total_usage, # Stitch results and update state sequentially tool_results = [] + stitched_names = [] for block in blocks: if block.id in skipped: tool_results.append( @@ -595,6 +646,7 @@ def _execute_job_code_tools_parallel(self, blocks, stream_manager, total_usage, self.current_yaml = stitch_job_code(self.current_yaml, matched_job_key, suggested_code) self.yaml_modified = True stitched = True + stitched_names.append(self._display_name_for_job(matched_job_key)) logger.info(f"Stitched code for job '{matched_job_key}' into current_yaml") self.subagent_results.append(subagent_result) @@ -611,6 +663,14 @@ def _execute_job_code_tools_parallel(self, blocks, stream_manager, total_usage, {"type": "tool_result", "tool_use_id": block.id, "content": tool_result} ) + # Settle the spinner with the steps that were actually applied (drop any + # that failed to stitch); nothing sent if none applied. One YAML send + # covers the whole batch, mirroring the one combined status. + if stitched_names: + self._send_yaml(stream_manager) + joined = ", ".join(f"\"{n}\"" for n in stitched_names) + self._send_settled(stream_manager, f"Wrote code for {joined}") + return tool_results def _tool_status_message(self, tool_use_block) -> str: @@ -626,7 +686,7 @@ def _tool_status_message(self, tool_use_block) -> str: if name == "call_workflow_agent": if self.current_yaml: - return "Editing workflow..." + return "Reviewing the workflow..." return "Building workflow outline..." if name == "call_job_code_agent": @@ -646,6 +706,38 @@ def _tool_status_message(self, tool_use_block) -> str: return f"Running {name}..." + def _settled_status_message(self, tool_use_block, yaml_before: str | None) -> str | None: + """Past-tense line that resolves the spinner for a finished tool call. + + Counterpart to _tool_status_message. For workflow edits the outcome is + read from whether the YAML actually changed: an unchanged workflow means + the agent only advised (or errored), so it settles to "Analyzed the + workflow" rather than claiming an edit. Returns None when there's + nothing worth persisting. Job-code settling is handled where the code is + stitched, since it depends on which steps were applied. + """ + name = tool_use_block.name + inputs = tool_use_block.input or {} + + if name == "call_workflow_agent": + if self.current_yaml == yaml_before: + return "Analyzed the workflow" + return "Edited workflow structure" if yaml_before else "Built workflow outline" + + if name == "search_documentation": + query = inputs.get("query") + return f"Searched documentation for \"{query}\"" if query else "Searched documentation" + + if name == "inspect_job_code": + job_keys = inputs.get("job_keys") or ([inputs["job_key"]] if inputs.get("job_key") else []) + names = [n for n in (self._display_name_for_job(k) for k in job_keys) if n] + if names: + joined = ", ".join(f"\"{n}\"" for n in names) + return f"Read code for {joined}" + return "Read code" + + return None + def _display_name_for_job(self, job_key: str | None) -> str | None: """Look up a human-readable display name for a job key. @@ -668,12 +760,8 @@ def _format_display_name(name: str) -> str: return name.replace("-", " ").replace("_", " ").title() def _extract_text(self, response): - """Extract text from response content.""" - text = "" - for block in response.content: - if block.type == "text": - text += block.text - return text + """Extract text from response content, concatenated as it was streamed.""" + return "".join(block.text for block in response.content if block.type == "text") def _build_system_prompt(self) -> list: """Build system prompt for planner with cache control.""" diff --git a/services/global_chat/router.py b/services/global_chat/router.py index 238cf4dc..db0bdcf3 100644 --- a/services/global_chat/router.py +++ b/services/global_chat/router.py @@ -40,6 +40,7 @@ class RouterResult: """Result from router or passthrough.""" response: str + response_segments: List[Dict] attachments: List[Dict] history: List[Dict] usage: Dict @@ -258,6 +259,7 @@ def _route_to_workflow_chat( return RouterResult( response=result["response"], + response_segments=[{"type": "text", "content": result["response"]}], attachments=attachments, history=result["history"].copy(), usage=total_usage, @@ -360,6 +362,7 @@ def _route_to_job_chat( return RouterResult( response=result["response"], + response_segments=[{"type": "text", "content": result["response"]}], attachments=attachments, history=result["history"].copy(), usage=total_usage, @@ -401,6 +404,7 @@ def _route_to_planner( return RouterResult( response=planner_result.response, + response_segments=planner_result.response_segments, attachments=planner_result.attachments, history=planner_result.history, usage=total_usage, diff --git a/services/global_chat/tests/unit/test_planner.py b/services/global_chat/tests/unit/test_planner.py index 469e10b3..369ec9a1 100644 --- a/services/global_chat/tests/unit/test_planner.py +++ b/services/global_chat/tests/unit/test_planner.py @@ -22,6 +22,7 @@ def make_planner() -> PlannerAgent: planner.current_yaml = WORKFLOW_YAML planner.yaml_modified = False planner.subagent_results = [] + planner._segments = [] planner.api_key = "test-key" planner._user = None planner._metrics_opt_in = None @@ -48,12 +49,18 @@ class StubStreamManager: def send_thinking(self, *_args: object, **_kwargs: object) -> None: pass + def send_changes(self, *_args: object, **_kwargs: object) -> None: + pass + + def send_status(self, *_args: object, **_kwargs: object) -> None: + pass + def test_inspect_job_code_accepts_multiple_keys() -> None: planner = make_planner() block = FakeToolUse("inspect_job_code", {"job_keys": ["fetch-patients", "missing-step"]}) - result = planner._execute_tool(block, empty_usage(), []) + result = planner._execute_tool(block, StubStreamManager(), empty_usage(), []) assert "get('/patients');" in result assert "No code found for job 'missing-step'" in result @@ -65,7 +72,7 @@ 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, empty_usage(), meta) + result = planner._execute_tool(block, StubStreamManager(), empty_usage(), meta) assert result.startswith("ERROR: The job code agent failed: boom") assert meta[0]["error"] == "boom" @@ -76,7 +83,7 @@ def test_workflow_agent_failure_returns_error_tool_result() -> None: block = FakeToolUse("call_workflow_agent", {"message": "add a step"}) with patch("global_chat.planner.call_workflow_agent", side_effect=RuntimeError("boom")): - result = planner._execute_tool(block, empty_usage(), []) + result = planner._execute_tool(block, StubStreamManager(), empty_usage(), []) assert result.startswith("ERROR: The workflow agent failed: boom") assert planner.current_yaml == WORKFLOW_YAML @@ -89,7 +96,7 @@ 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, empty_usage(), []) + result = planner._execute_tool(block, StubStreamManager(), empty_usage(), []) assert "NOT added to the workflow" in result assert "stitched into the workflow" not in result @@ -104,7 +111,7 @@ def test_workflow_agent_yaml_response_updates_structure_view() -> None: subagent_result = {"response": "Added the step.", "response_yaml": new_yaml, "usage": empty_usage()} with patch("global_chat.planner.call_workflow_agent", return_value=subagent_result): - result = planner._execute_tool(block, empty_usage(), []) + result = planner._execute_tool(block, StubStreamManager(), empty_usage(), []) assert "Updated workflow structure:" in result assert "new-step" in result @@ -118,7 +125,7 @@ def test_workflow_agent_without_yaml_reports_no_change() -> None: subagent_result = {"response": "Which DHIS2 instance?", "response_yaml": None, "usage": empty_usage()} with patch("global_chat.planner.call_workflow_agent", return_value=subagent_result): - result = planner._execute_tool(block, empty_usage(), []) + result = planner._execute_tool(block, StubStreamManager(), empty_usage(), []) assert "[No workflow changes were made — no YAML was produced.]" in result assert "Updated workflow structure:" not in result diff --git a/services/streaming_util.py b/services/streaming_util.py index c95ec700..08cb6dce 100644 --- a/services/streaming_util.py +++ b/services/streaming_util.py @@ -310,6 +310,24 @@ def send_changes(self, changes_data: dict[str, Any]) -> None: self._close_open_blocks() self._emit_event('changes', changes_data) + def send_status(self, content: str) -> None: + """ + Send a completed-action status ("Edited workflow structure") as a + custom `status` SSE event. + + This is deliberately a different event type from the transient + spinners sent via send_thinking: thinking events are live progress + that the client replaces and never persists, while `status` events + are durable facts about what happened, which the client keeps. The + payload matches the `response_segments` entry shape so the client + can render live events and reloaded segments with the same code. + """ + if not self.stream_started: + self.start_stream() + + self._close_open_blocks() + self._emit_event('status', {"type": "status", "content": content}) + def end_stream(self, stop_reason: str = "end_turn") -> None: """ End the stream by closing all open blocks and sending final events. From 771a2108c18e35e10cbf32dde916fa3f4395681f Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:18:21 +0100 Subject: [PATCH 2/5] Bump the python-minor-patch group across 1 directory with 8 updates (#601) Bumps the python-minor-patch group with 8 updates in the / directory: | Package | From | To | | --- | --- | --- | | [openai](https://github.com/openai/openai-python) | `2.45.0` | `2.48.0` | | [anthropic](https://github.com/anthropics/anthropic-sdk-python) | `0.116.0` | `0.119.0` | | [langchain-core](https://github.com/langchain-ai/langchain) | `1.4.9` | `1.5.1` | | [langchain-openai](https://github.com/langchain-ai/langchain) | `1.3.4` | `1.4.1` | | [sentry-sdk](https://github.com/getsentry/sentry-python) | `2.64.0` | `2.66.1` | | [langfuse](https://github.com/langfuse/langfuse) | `4.14.0` | `4.14.1` | | [opentelemetry-instrumentation-threading](https://github.com/open-telemetry/opentelemetry-python-contrib) | `0.64b0` | `0.65b0` | | [ruff](https://github.com/astral-sh/ruff) | `0.15.21` | `0.16.0` | Updates `openai` from 2.45.0 to 2.48.0 - [Release notes](https://github.com/openai/openai-python/releases) - [Changelog](https://github.com/openai/openai-python/blob/main/CHANGELOG.md) - [Commits](https://github.com/openai/openai-python/compare/v2.45.0...v2.48.0) Updates `anthropic` from 0.116.0 to 0.119.0 - [Release notes](https://github.com/anthropics/anthropic-sdk-python/releases) - [Changelog](https://github.com/anthropics/anthropic-sdk-python/blob/main/CHANGELOG.md) - [Commits](https://github.com/anthropics/anthropic-sdk-python/compare/v0.116.0...v0.119.0) Updates `langchain-core` from 1.4.9 to 1.5.1 - [Release notes](https://github.com/langchain-ai/langchain/releases) - [Commits](https://github.com/langchain-ai/langchain/compare/langchain-core==1.4.9...langchain-core==1.5.1) Updates `langchain-openai` from 1.3.4 to 1.4.1 - [Release notes](https://github.com/langchain-ai/langchain/releases) - [Commits](https://github.com/langchain-ai/langchain/compare/langchain-openai==1.3.4...langchain-openai==1.4.1) Updates `sentry-sdk` from 2.64.0 to 2.66.1 - [Release notes](https://github.com/getsentry/sentry-python/releases) - [Changelog](https://github.com/getsentry/sentry-python/blob/master/CHANGELOG.md) - [Commits](https://github.com/getsentry/sentry-python/compare/2.64.0...2.66.1) Updates `langfuse` from 4.14.0 to 4.14.1 - [Release notes](https://github.com/langfuse/langfuse/releases) - [Commits](https://github.com/langfuse/langfuse/commits) Updates `opentelemetry-instrumentation-threading` from 0.64b0 to 0.65b0 - [Release notes](https://github.com/open-telemetry/opentelemetry-python-contrib/releases) - [Changelog](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/CHANGELOG.md) - [Commits](https://github.com/open-telemetry/opentelemetry-python-contrib/commits) Updates `ruff` from 0.15.21 to 0.16.0 - [Release notes](https://github.com/astral-sh/ruff/releases) - [Changelog](https://github.com/astral-sh/ruff/blob/main/CHANGELOG.md) - [Commits](https://github.com/astral-sh/ruff/compare/0.15.21...0.16.0) --- updated-dependencies: - dependency-name: openai dependency-version: 2.48.0 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: python-minor-patch - dependency-name: anthropic dependency-version: 0.119.0 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: python-minor-patch - dependency-name: langchain-core dependency-version: 1.5.1 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: python-minor-patch - dependency-name: langchain-openai dependency-version: 1.4.1 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: python-minor-patch - dependency-name: sentry-sdk dependency-version: 2.66.1 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: python-minor-patch - dependency-name: langfuse dependency-version: 4.14.1 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: python-minor-patch - dependency-name: opentelemetry-instrumentation-threading dependency-version: 0.65b0 dependency-type: direct:production dependency-group: python-minor-patch - dependency-name: ruff dependency-version: 0.16.0 dependency-type: direct:development update-type: version-update:semver-minor dependency-group: python-minor-patch ... Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> --- poetry.lock | 156 +++++++++++++++++++++++++------------------------ pyproject.toml | 16 ++--- 2 files changed, 87 insertions(+), 85 deletions(-) diff --git a/poetry.lock b/poetry.lock index 800de0e0..c413ba5d 100644 --- a/poetry.lock +++ b/poetry.lock @@ -199,14 +199,14 @@ files = [ [[package]] name = "anthropic" -version = "0.116.0" +version = "0.119.0" description = "The official Python library for the anthropic API" optional = false python-versions = ">=3.9" groups = ["main"] files = [ - {file = "anthropic-0.116.0-py3-none-any.whl", hash = "sha256:6c0a7698e8d652455da3499978279bb2588c7264d0a35be3666009a4258c8256"}, - {file = "anthropic-0.116.0.tar.gz", hash = "sha256:5fc248fbb9fe03ef686f8a774f81586bca31a043260aab88b387ea3660f4a396"}, + {file = "anthropic-0.119.0-py3-none-any.whl", hash = "sha256:2b39107077489d19d7c66000e9d9263293d431550fcf6b543ead785c6ac85962"}, + {file = "anthropic-0.119.0.tar.gz", hash = "sha256:40a81b094ae2a0056a77771257943b19bd46edce1326e22f85c2794492097f0a"}, ] [package.dependencies] @@ -220,9 +220,10 @@ sniffio = "*" typing-extensions = ">=4.14,<5" [package.extras] -aiohttp = ["aiohttp", "httpx-aiohttp (>=0.1.9)"] +aiohttp = ["aiohttp", "httpx-aiohttp (>=0.1.9,<1)"] aws = ["boto3 (>=1.28.57)", "botocore (>=1.31.57)"] bedrock = ["boto3 (>=1.28.57)", "botocore (>=1.31.57)"] +google-cloud = ["google-auth[requests] (>=2,<3)"] mcp = ["mcp (>=1.0) ; python_version >= \"3.10\""] vertex = ["google-auth[requests] (>=2,<3)"] webhooks = ["standardwebhooks (>=1.0.1,<2)"] @@ -1045,14 +1046,14 @@ tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<10.0.0" [[package]] name = "langchain-core" -version = "1.4.9" +version = "1.5.1" description = "Building applications with LLMs through composability" optional = false python-versions = "<4.0.0,>=3.10.0" groups = ["main"] files = [ - {file = "langchain_core-1.4.9-py3-none-any.whl", hash = "sha256:28e3909e2a10cc81504952d795ac0a9e014c0018121ef89d48dd396fa09ec624"}, - {file = "langchain_core-1.4.9.tar.gz", hash = "sha256:f8078901145bed0466755277500a5a22822a7b628808c4c0a28d4fc88895fcf2"}, + {file = "langchain_core-1.5.1-py3-none-any.whl", hash = "sha256:c5ec8f51dd05124f950c9afd0fd8bb3f7be4e405eeb868d481b2f8ff652cb9d2"}, + {file = "langchain_core-1.5.1.tar.gz", hash = "sha256:b0df382704c6403c1e0c9603415bce09290455d4aeeb38f350d15f78ca597483"}, ] [package.dependencies] @@ -1068,19 +1069,19 @@ uuid-utils = ">=0.12.0,<1.0" [[package]] name = "langchain-openai" -version = "1.3.4" +version = "1.4.1" description = "An integration package connecting OpenAI and LangChain" optional = false python-versions = "<4.0.0,>=3.10.0" groups = ["main"] files = [ - {file = "langchain_openai-1.3.4-py3-none-any.whl", hash = "sha256:3241b8392b29c1af233b902b7d9a84bfc5fe26ccb210a7febdfc3972af7e5771"}, - {file = "langchain_openai-1.3.4.tar.gz", hash = "sha256:d888d5f39c2a8c3d0d8aa88f5cf50e58a8e7d242f3f15e39422add520eec8e31"}, + {file = "langchain_openai-1.4.1-py3-none-any.whl", hash = "sha256:8528bb34cc78fdfd2d895573c7917f9441cbb82db5f18ae0e6b3b75d95bdefb3"}, + {file = "langchain_openai-1.4.1.tar.gz", hash = "sha256:6d16be615d997db80294731b8e768783f1fb8e0313668e64acd50cd68acbad20"}, ] [package.dependencies] -langchain-core = ">=1.4.9,<2.0.0" -openai = ">=2.26.0,<3.0.0" +langchain-core = ">=1.5.1,<2.0.0" +openai = ">=2.45.0,<3.0.0" tiktoken = ">=0.7.0,<1.0.0" [[package]] @@ -1136,14 +1137,14 @@ langchain-core = ">=1.2.31,<2.0.0" [[package]] name = "langfuse" -version = "4.14.0" -description = "A client library for accessing langfuse" +version = "4.14.1" +description = "Langfuse Python SDK - LLM observability/tracing, datasets, experiments, LLM-as-a-judge evaluation, and prompt management" optional = false python-versions = "<4.0,>=3.10" groups = ["main"] files = [ - {file = "langfuse-4.14.0-py3-none-any.whl", hash = "sha256:632b74abfa14d9ad3dccbe3322bf7e11b0dce9be0de451b0d134db9bff246d44"}, - {file = "langfuse-4.14.0.tar.gz", hash = "sha256:31b875c8d09eee39c558584b9424bbd5ed014965c5944c4528a37947ae4b7787"}, + {file = "langfuse-4.14.1-py3-none-any.whl", hash = "sha256:07d19f16338b8e21f8e5996b7e6c3ed150ee582fbaa6275ac9eeea297093f4be"}, + {file = "langfuse-4.14.1.tar.gz", hash = "sha256:576641820ae79aeca71453c8eff3c8c35d53f7e41729c918f59eb81f7499faf9"}, ] [package.dependencies] @@ -1421,14 +1422,14 @@ files = [ [[package]] name = "openai" -version = "2.45.0" +version = "2.48.0" description = "The official Python library for the openai API" optional = false python-versions = ">=3.9" groups = ["main"] files = [ - {file = "openai-2.45.0-py3-none-any.whl", hash = "sha256:5df105f5f8c9b711fcb9d06d2d3888cebc82506db216484c14a4e53cdf651777"}, - {file = "openai-2.45.0.tar.gz", hash = "sha256:10d34ca9c5643bce775852fddbfc172505cb1d4de1ccd101696c3ecff358765d"}, + {file = "openai-2.48.0-py3-none-any.whl", hash = "sha256:c98df30aaaf93c51979f64d3e7c5b76464f8be0173368266229eb8fe6bd30f2c"}, + {file = "openai-2.48.0.tar.gz", hash = "sha256:231b1e7661dda14574986c2f71451e9d584b7fe69e0ee6480e12ed090b48fc16"}, ] [package.dependencies] @@ -1442,22 +1443,23 @@ tqdm = ">4" typing-extensions = ">=4.14,<5" [package.extras] -aiohttp = ["aiohttp", "httpx-aiohttp (>=0.1.9)"] +aiohttp = ["aiohttp (>=3.14.1) ; python_version >= \"3.10\"", "httpx-aiohttp (>=0.1.9) ; python_version >= \"3.10\""] bedrock = ["botocore (>=1.40.0,<1.43) ; python_version < \"3.10\"", "botocore (>=1.40.0,<2) ; python_version >= \"3.10\""] datalib = ["numpy (>=1)", "pandas (>=1.2.3)", "pandas-stubs (>=1.1.0.11)"] +httpx2 = ["anyio (>=4.10.0,<5) ; python_version >= \"3.10\"", "httpx (>=0.25.1,<1) ; python_version >= \"3.10\"", "httpx2 (>=2.7.0,<3) ; python_version >= \"3.10\""] realtime = ["websockets (>=13,<16)"] voice-helpers = ["numpy (>=2.0.2)", "sounddevice (>=0.5.1)"] [[package]] name = "opentelemetry-api" -version = "1.43.0" +version = "1.44.0" description = "OpenTelemetry Python API" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_api-1.43.0-py3-none-any.whl", hash = "sha256:20acf45e9b21851926835292e4045d290acade1edd2ff3de86d2f069687ba1fd"}, - {file = "opentelemetry_api-1.43.0.tar.gz", hash = "sha256:107d0d03857ea8fc7c5fcbbbd83f800c281f0d560553d61c1d675fccfd1761c1"}, + {file = "opentelemetry_api-1.44.0-py3-none-any.whl", hash = "sha256:94b98c893a91b88657eaac1e3ba89618cdb85be6918196705354f34728b2cdef"}, + {file = "opentelemetry_api-1.44.0.tar.gz", hash = "sha256:67647e5e9566edcf421166fdf022b3537f818635daa852b289e34604dc6fb33a"}, ] [package.dependencies] @@ -1465,37 +1467,37 @@ typing-extensions = ">=4.5.0" [[package]] name = "opentelemetry-exporter-otlp-proto-common" -version = "1.43.0" +version = "1.44.0" description = "OpenTelemetry Protobuf encoding" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_exporter_otlp_proto_common-1.43.0-py3-none-any.whl", hash = "sha256:123c3f9cc87218562490c63b36f497bf3a722faf174a515d1443f31ababa6264"}, - {file = "opentelemetry_exporter_otlp_proto_common-1.43.0.tar.gz", hash = "sha256:c4e32ba6d6b13bdb2b8f6764c4fd28d00192826561aa04f6d14eedfce7ac076f"}, + {file = "opentelemetry_exporter_otlp_proto_common-1.44.0-py3-none-any.whl", hash = "sha256:9a9fe61bba73d802904bc989f1d6b4a7b1ee40f06c40e98d6f85af65aaebb694"}, + {file = "opentelemetry_exporter_otlp_proto_common-1.44.0.tar.gz", hash = "sha256:dc87a5a5bc58f149a56d1547e4691588fa12994cdc3bc039a694ccb3375862ac"}, ] [package.dependencies] -opentelemetry-proto = "1.43.0" +opentelemetry-proto = "1.44.0" [[package]] name = "opentelemetry-exporter-otlp-proto-http" -version = "1.43.0" +version = "1.44.0" description = "OpenTelemetry Collector Protobuf over HTTP Exporter" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_exporter_otlp_proto_http-1.43.0-py3-none-any.whl", hash = "sha256:647f603aa8efdbdb4dbff842e0729d0406a6fff26b295a72d3d60e7d963b2610"}, - {file = "opentelemetry_exporter_otlp_proto_http-1.43.0.tar.gz", hash = "sha256:fa8a42bb7d00ee5391f4c0b04d8e6a46c03caa437903296ab73a81dc11ba118f"}, + {file = "opentelemetry_exporter_otlp_proto_http-1.44.0-py3-none-any.whl", hash = "sha256:838592fce774c1c8bb7b9a0a7facbfa82e17be5a8a4e94cef10cb84ae026bae3"}, + {file = "opentelemetry_exporter_otlp_proto_http-1.44.0.tar.gz", hash = "sha256:c633d7270ad6b57cd4cfbe8b0007a9e2e7c0cb50bd6c50fe2a7b245f721a09d8"}, ] [package.dependencies] googleapis-common-protos = ">=1.52,<2.0" opentelemetry-api = ">=1.15,<2.0" -opentelemetry-exporter-otlp-proto-common = "1.43.0" -opentelemetry-proto = "1.43.0" -opentelemetry-sdk = ">=1.43.0,<1.44.0" +opentelemetry-exporter-otlp-proto-common = "1.44.0" +opentelemetry-proto = "1.44.0" +opentelemetry-sdk = ">=1.44.0,<1.45.0" requests = ">=2.7,<3.0" typing-extensions = ">=4.5.0" @@ -1504,19 +1506,19 @@ gcp-auth = ["opentelemetry-exporter-credential-provider-gcp (>=0.59b0)"] [[package]] name = "opentelemetry-instrumentation" -version = "0.64b0" +version = "0.65b0" description = "Instrumentation Tools & Auto Instrumentation for OpenTelemetry Python" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_instrumentation-0.64b0-py3-none-any.whl", hash = "sha256:133ab7ffca796557aec059bf6be3190a34b6dea987f25be3d9409e230cbdad8b"}, - {file = "opentelemetry_instrumentation-0.64b0.tar.gz", hash = "sha256:b47d528dead6271d7743114417eb67fc915bd9258111c48dbf9a4951d2efa88d"}, + {file = "opentelemetry_instrumentation-0.65b0-py3-none-any.whl", hash = "sha256:ea967a72b9939b5fcfdad572753b4306c59dcb99e3f382d95dae04286805e137"}, + {file = "opentelemetry_instrumentation-0.65b0.tar.gz", hash = "sha256:071d9d9eced9bd6460444ec3b0c77229870ed05a881c22c84fdede58e4eed09b"}, ] [package.dependencies] opentelemetry-api = ">=1.4,<2.0" -opentelemetry-semantic-conventions = "0.64b0" +opentelemetry-semantic-conventions = "0.65b0" packaging = ">=18.0" wrapt = ">=1.0.0,<3.0.0" @@ -1543,31 +1545,31 @@ instruments = ["anthropic"] [[package]] name = "opentelemetry-instrumentation-threading" -version = "0.64b0" +version = "0.65b0" description = "Thread context propagation support for OpenTelemetry" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_instrumentation_threading-0.64b0-py3-none-any.whl", hash = "sha256:a285ffa750a958f7d368e947f5679a0214d588242cbffba0f5934cb02e9a17f4"}, - {file = "opentelemetry_instrumentation_threading-0.64b0.tar.gz", hash = "sha256:0a07d7329f69dfae5036a7cb184f502c8b91cb0538012f5304bd32ffe9ade451"}, + {file = "opentelemetry_instrumentation_threading-0.65b0-py3-none-any.whl", hash = "sha256:d8a1a1f35418a32769d469ef2d7e8401935e097a8553abcfba833da9c74736ce"}, + {file = "opentelemetry_instrumentation_threading-0.65b0.tar.gz", hash = "sha256:aefd23eb16c5e7a7c6c6eacdcb6c6f269ed8ed3a1b458bf5dd555b878bf58937"}, ] [package.dependencies] opentelemetry-api = ">=1.12,<2.0" -opentelemetry-instrumentation = "0.64b0" +opentelemetry-instrumentation = "0.65b0" wrapt = ">=1.0.0,<3.0.0" [[package]] name = "opentelemetry-proto" -version = "1.43.0" +version = "1.44.0" description = "OpenTelemetry Python Proto" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_proto-1.43.0-py3-none-any.whl", hash = "sha256:c58f1f7ef84bc7dc2834016c0c37fe0081dde7ca9f6339be1970fbf9cdaaa90d"}, - {file = "opentelemetry_proto-1.43.0.tar.gz", hash = "sha256:224778df17e1f3fafeaaa21d874236ca5f6ffc2f86e0899298ec7351aac27924"}, + {file = "opentelemetry_proto-1.44.0-py3-none-any.whl", hash = "sha256:898b155a0e1557afd867478fb6158e8122a46329ca0bb8dc53cc55e98f017f56"}, + {file = "opentelemetry_proto-1.44.0.tar.gz", hash = "sha256:c547a79c2f8c0c515d31509154682e5921c7cfd5ca67b70e1f9266e2c3e103f3"}, ] [package.dependencies] @@ -1575,38 +1577,38 @@ protobuf = ">=5.0,<8.0" [[package]] name = "opentelemetry-sdk" -version = "1.43.0" +version = "1.44.0" description = "OpenTelemetry Python SDK" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_sdk-1.43.0-py3-none-any.whl", hash = "sha256:d1323a547c1ce69d6a069a17a44b7da82bb8b332051ecb074041f87642c86823"}, - {file = "opentelemetry_sdk-1.43.0.tar.gz", hash = "sha256:d8187c81c162df9913e4003dd6485f7390d9a24fc17026ec7387b8b8218b08e9"}, + {file = "opentelemetry_sdk-1.44.0-py3-none-any.whl", hash = "sha256:df081c4c6bcfdb1211e3e86140376792643128a25f8d72d1d27675936e7e96ad"}, + {file = "opentelemetry_sdk-1.44.0.tar.gz", hash = "sha256:cebe7f65dc12f26ead75c6064de12fd2a9052e5060c0272d402cfa203aae123b"}, ] [package.dependencies] -opentelemetry-api = "1.43.0" -opentelemetry-semantic-conventions = "0.64b0" +opentelemetry-api = "1.44.0" +opentelemetry-semantic-conventions = "0.65b0" typing-extensions = ">=4.5.0" [package.extras] -file-configuration = ["jsonschema (>=4.0)", "pyyaml (>=6.0)"] +file-configuration = ["opentelemetry-configuration (==0.65b0)"] [[package]] name = "opentelemetry-semantic-conventions" -version = "0.64b0" +version = "0.65b0" description = "OpenTelemetry Semantic Conventions" optional = false python-versions = ">=3.10" groups = ["main"] files = [ - {file = "opentelemetry_semantic_conventions-0.64b0-py3-none-any.whl", hash = "sha256:ea77e85e354b8f604ddbe5f3d9135216f982fa4d77e5859ac30f6d8a50505aa6"}, - {file = "opentelemetry_semantic_conventions-0.64b0.tar.gz", hash = "sha256:72f76fb2d1582d9d033dd1fcd84532e961e6ff3d90d24ba6fabc72975a83864c"}, + {file = "opentelemetry_semantic_conventions-0.65b0-py3-none-any.whl", hash = "sha256:1cacde7b0ad306f84c5ef08c3dbe1bbaf20165bba6f8bff43b670e555a086bcb"}, + {file = "opentelemetry_semantic_conventions-0.65b0.tar.gz", hash = "sha256:f9b2b81e9d5b64f11bc952075e7e9c7fb0aab075c7fd1c46d597f1b919852d60"}, ] [package.dependencies] -opentelemetry-api = "1.43.0" +opentelemetry-api = "1.44.0" typing-extensions = ">=4.5.0" [[package]] @@ -2513,42 +2515,42 @@ requests = ">=2.0.1,<3.0.0" [[package]] name = "ruff" -version = "0.15.21" +version = "0.16.0" description = "An extremely fast Python linter and code formatter, written in Rust." optional = false python-versions = ">=3.7" groups = ["dev"] files = [ - {file = "ruff-0.15.21-py3-none-linux_armv6l.whl", hash = "sha256:63ea0e965e5d73c90e95b2434beeafc70820536717f561b32ab6e777cb9bdf5d"}, - {file = "ruff-0.15.21-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:0f212c5d7d54c01bbfe6dcab02b724a39300f3e34ed7acbe995ccb320a2c58bd"}, - {file = "ruff-0.15.21-py3-none-macosx_11_0_arm64.whl", hash = "sha256:e6312e41bc96791299614995ea3a977c5857c3b5662b1ecef6755b02b87cb646"}, - {file = "ruff-0.15.21-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:01d65b4831c6b2a4ba8ee6faa84049d44d982b7a706e622c4094c509e51673be"}, - {file = "ruff-0.15.21-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:2c5a913a589120ce67933d5d05fd6ddbcc2481c6a054980ee767f7414c72b4fd"}, - {file = "ruff-0.15.21-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:5ef04b681d02ad4dc9620f00f83ac5c22f652d0e9a9cfe431d219b16ad5ccc41"}, - {file = "ruff-0.15.21-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:16d090c0740916594157e75b80d666eab8e78083b39b3b0e1d698f4670a17b86"}, - {file = "ruff-0.15.21-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3a10e74757dd65004d779b73e2f3c5210156d9980b41224d50d2ebcf1db51e67"}, - {file = "ruff-0.15.21-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bab0905d2f29e0d9fbc3c373ed23db0095edaa3f71f1f4f519ec15134d9e85c8"}, - {file = "ruff-0.15.21-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:00eca240af5789fec6fe7df74c088cc1f9644ed83027113468efba7c92b94075"}, - {file = "ruff-0.15.21-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:262ab31557a75141325e32d3357f3597645a7f084e732b6b054dde428ecd9341"}, - {file = "ruff-0.15.21-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:659c4e7a4212f83306045ec7c5e5a356d16d9a6ef4ae0c7a4d872914fc655d9d"}, - {file = "ruff-0.15.21-py3-none-musllinux_1_2_i686.whl", hash = "sha256:9e866eab611a5f959d36df2d10e446973a3610bc42b0c15b31dc27977d59c233"}, - {file = "ruff-0.15.21-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:e89bc93c0d3803ba870b55c29671bad9dc6d94bb1eb181b056b52eb05b52854f"}, - {file = "ruff-0.15.21-py3-none-win32.whl", hash = "sha256:01f8d5be84823c172b389e123174f781f9daf86d6c58719d603f941932195cdd"}, - {file = "ruff-0.15.21-py3-none-win_amd64.whl", hash = "sha256:d4b8d9a2f0f12b816b50447f6eccb9f4bb01a6b82c86b50fb3b5354b458dc6d3"}, - {file = "ruff-0.15.21-py3-none-win_arm64.whl", hash = "sha256:6e83115d4b9377c1cbc13abf0e051f069fab0ef815ea0504a8a008cee24dd0a8"}, - {file = "ruff-0.15.21.tar.gz", hash = "sha256:d0cfc841c572283c36548f82664a54ce6565567f1b0d5b4cf2caac693d8b7500"}, + {file = "ruff-0.16.0-py3-none-linux_armv6l.whl", hash = "sha256:e5115729eb08c585e5121978ba5d5b60caeae394ce21b9fb5e6cd33a1c6c9b1e"}, + {file = "ruff-0.16.0-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:3c954b1d580bfa035b41654f7858cc7e71d5fc3ac5b723dd62bd9133830ed522"}, + {file = "ruff-0.16.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:e01c21d10eb1b29f47b7454e1f4056db9a3f0260c646aa88457c610291db9f81"}, + {file = "ruff-0.16.0-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6e364e5ed22ed8dc05082fd78e35308618260907ac2d3c1d637b2e682415b6c9"}, + {file = "ruff-0.16.0-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:d327b8fc113a1d4421a04f3839d3752057c8dd1ee320223a6f3f52d04ada462a"}, + {file = "ruff-0.16.0-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a9b50c55e263103586b3dcf5f73d479eb8cb5fdb6098fec59a62891dab653717"}, + {file = "ruff-0.16.0-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:0ff4a79ce3ec0172f3241943835de1c4cb4e2dcd07f0f8c2d02603dbbbee4b17"}, + {file = "ruff-0.16.0-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:e95c448fca1fb2a18372a9440926c5a6ee789639bb975c72e7ae6d0b04218ab4"}, + {file = "ruff-0.16.0-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4f11a8d11010301d0a398a2fdef67691feca7294da6aef55e2150e8fa2cd520b"}, + {file = "ruff-0.16.0-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:48044c678e9cb8698246c99b14aaccfa6601dea7379eb48a6f8f73f7a6d86cd0"}, + {file = "ruff-0.16.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:7aa0959bad8eb8bef50340154fc9b58678dae31fa4293afa38b44b6e552c0213"}, + {file = "ruff-0.16.0-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:28ea2b7df8ebf7f9da6b7d47b230ab48f387c0a29be3b474c4d0740e197bb9af"}, + {file = "ruff-0.16.0-py3-none-musllinux_1_2_i686.whl", hash = "sha256:33a3dfac8c35f81498dea9181bccc2f4c4bc8f1521a1dd9406e77643e0f0fb09"}, + {file = "ruff-0.16.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:a5237a0bda500d30d81b8e07a6973a5cbc772864cbf746ae2f4e8a2e01c9f4ed"}, + {file = "ruff-0.16.0-py3-none-win32.whl", hash = "sha256:7fab76fa065c873f41ff744347c6e77bcc3dfec4bcc754dc26b63d23c0f7f5fb"}, + {file = "ruff-0.16.0-py3-none-win_amd64.whl", hash = "sha256:429c117f022bf481fabd9d551e7a3952b24c65e6ef44337ea09d90bebef14472"}, + {file = "ruff-0.16.0-py3-none-win_arm64.whl", hash = "sha256:14296fedcd2705c77ab8235439278bbb38f285cf7da5528b00b3e330c3d4872d"}, + {file = "ruff-0.16.0.tar.gz", hash = "sha256:e460aafd5495ec89efaa6ced2e4a9a581116451e1c88b9d37ef497e0f8e93982"}, ] [[package]] name = "sentry-sdk" -version = "2.64.0" +version = "2.66.1" description = "Python client for Sentry (https://sentry.io)" optional = false python-versions = ">=3.6" groups = ["main"] files = [ - {file = "sentry_sdk-2.64.0-py3-none-any.whl", hash = "sha256:715ea91ca860a819e8d8a50a7bde3a80d0df3b4ed7b6660a20fb9a2d084188f1"}, - {file = "sentry_sdk-2.64.0.tar.gz", hash = "sha256:68be2c29e14ae310f8a39e1a79916b6d85c6cb41dcce789d14ff05fe293e4c55"}, + {file = "sentry_sdk-2.66.1-py3-none-any.whl", hash = "sha256:86002793161d9a95ef04bdd8d442e9bfece5d989b755f05d6360215094a7aff6"}, + {file = "sentry_sdk-2.66.1.tar.gz", hash = "sha256:f882fb08710c5f8bfc603aafa3e901b384009a19cc3f76a572b863392ee81cdc"}, ] [package.dependencies] @@ -3623,4 +3625,4 @@ cffi = ["cffi (>=1.17,<2.0) ; platform_python_implementation != \"PyPy\" and pyt [metadata] lock-version = "2.1" python-versions = "3.11.*" -content-hash = "1987d4f8e6869a49e46e2011f8d4557e10b566fa878f6f87183a67acf75408a8" +content-hash = "6551e8fae076b04ea022e47a09446e6573992a90751c4ce633f9e2e0d73f035e" diff --git a/pyproject.toml b/pyproject.toml index 896548e5..f67ccac0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,29 +10,29 @@ requires-poetry = ">=2.3.2" [tool.poetry.dependencies] python = "3.11.*" -openai = "^2.45" +openai = "^2.48" python-dotenv = "^1.2.2" -anthropic = "^0.116.0" +anthropic = "^0.119.0" langchain-pinecone = "^0.2.13" -langchain-core = "^1.4" +langchain-core = "^1.5" langchain-community = "^0.4.2" -langchain-openai = "^1.3" +langchain-openai = "^1.4" langchain-text-splitters = "^1.1" nltk = "^3.10.0" pytest = "^9.1.1" -sentry-sdk = "^2.64.0" +sentry-sdk = "^2.66.1" psycopg2-binary = "^2.9.10" -langfuse = "^4.14.0" +langfuse = "^4.14.1" opentelemetry-instrumentation-anthropic = "^0.62.1" -opentelemetry-instrumentation-threading = "0.64b0" +opentelemetry-instrumentation-threading = "0.65b0" [tool.poetry.group.dev] optional = false [tool.poetry.group.dev.dependencies] pytest = "^9.1.1" -ruff = "^0.15.21" +ruff = "^0.16.0" [build-system] requires = ["poetry-core"] From 3f8acdc9156b2525f2eaed1bb8758485427b2ae7 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:18:48 +0100 Subject: [PATCH 3/5] Bump actions/setup-python from 6 to 7 (#600) Bumps [actions/setup-python](https://github.com/actions/setup-python) from 6 to 7. - [Release notes](https://github.com/actions/setup-python/releases) - [Commits](https://github.com/actions/setup-python/compare/v6...v7) --- updated-dependencies: - dependency-name: actions/setup-python dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> --- .github/workflows/unit-tests.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/unit-tests.yaml b/.github/workflows/unit-tests.yaml index d26646ba..5fa2788f 100644 --- a/.github/workflows/unit-tests.yaml +++ b/.github/workflows/unit-tests.yaml @@ -17,7 +17,7 @@ jobs: run: pipx install poetry==2.3.2 - name: Set up Python 3.11 - uses: actions/setup-python@v6 + uses: actions/setup-python@v7 with: python-version: "3.11" cache: poetry @@ -66,7 +66,7 @@ jobs: run: pipx install poetry==2.3.2 - name: Set up Python 3.11 - uses: actions/setup-python@v6 + uses: actions/setup-python@v7 with: python-version: "3.11" cache: poetry From 6c322ae5bc8e58bbe36e6b76a6f1181f254a2002 Mon Sep 17 00:00:00 2001 From: Hanna Paasivirta Date: Thu, 30 Jul 2026 17:52:49 +0100 Subject: [PATCH 4/5] Handle routing failures (#591) * add subagent mode and inspect tool * edits and tests * udpate subagent prompt * add navigation info * add langfuse tracing * rename inspect_workflow tool to edit_workflow * changeset * resolve conflict * update test fixtures * add coherence instruction * fix prompt for existing workflows --------- Co-authored-by: Joe Clark --- .changeset/five-bags-rule.md | 6 + services/global_chat/planner.py | 27 +- services/global_chat/prompts.yaml | 22 +- services/global_chat/router.py | 97 +++++- services/global_chat/subagent_caller.py | 2 +- .../job_code/test_canvas_code_request.md | 112 +++++++ .../test_question_reading_another_step.md | 114 +++++++ .../test_rest_to_rest_sync_with_cron.md | 92 ++++++ .../global_chat/tests/integration/__init__.py | 0 .../tests/integration/test_handover.py | 136 ++++++++ .../global_chat/tests/unit/test_router.py | 103 +++++- .../global_chat/tests/unit/test_yaml_utils.py | 68 ++++ .../global_chat/tools/tool_definitions.py | 21 +- services/global_chat/yaml_utils.py | 123 -------- services/job_chat/job_chat.py | 297 +++++++++++++----- services/job_chat/prompt.py | 100 +++++- .../tests/unit/test_subagent_prompt.py | 77 +++++ services/workflow_chat/gen_project_prompt.py | 25 +- .../workflow_chat/gen_project_prompts.yaml | 12 + .../tests/unit/client/test_handover.py | 45 +++ .../unit/gen_project/test_prompt_build.py | 30 ++ services/workflow_chat/workflow_chat.py | 101 +++++- services/yaml_utils.py | 202 ++++++++++++ 23 files changed, 1548 insertions(+), 264 deletions(-) create mode 100644 .changeset/five-bags-rule.md create mode 100644 services/global_chat/tests/acceptance/job_code/test_canvas_code_request.md create mode 100644 services/global_chat/tests/acceptance/job_code/test_question_reading_another_step.md create mode 100644 services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md create mode 100644 services/global_chat/tests/integration/__init__.py create mode 100644 services/global_chat/tests/integration/test_handover.py create mode 100644 services/global_chat/tests/unit/test_yaml_utils.py delete mode 100644 services/global_chat/yaml_utils.py create mode 100644 services/job_chat/tests/unit/test_subagent_prompt.py create mode 100644 services/workflow_chat/tests/unit/client/test_handover.py create mode 100644 services/yaml_utils.py diff --git a/.changeset/five-bags-rule.md b/.changeset/five-bags-rule.md new file mode 100644 index 00000000..d8c883b6 --- /dev/null +++ b/.changeset/five-bags-rule.md @@ -0,0 +1,6 @@ +--- +"apollo": patch +--- + +global_chat: enable subagents to pull missing context, recovering from routing +errors diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index 5fec69ff..9e603ac1 100644 --- a/services/global_chat/planner.py +++ b/services/global_chat/planner.py @@ -26,7 +26,7 @@ from global_chat.config_loader import ConfigLoader from models import resolve_model from global_chat.tools.tool_definitions import TOOL_DEFINITIONS -from global_chat.yaml_utils import stitch_job_code, redact_job_bodies, find_job_in_yaml, get_step_name_from_page +from yaml_utils import stitch_job_code, redact_job_bodies, find_job_in_yaml, get_step_name_from_page, inspect_job_code from tools.search_documentation.search_documentation import search_documentation_tool from global_chat.subagent_caller import call_workflow_agent, call_job_agent, format_subagent_result_for_llm @@ -81,6 +81,7 @@ def run( stream: bool, user: Optional[Dict] = None, metrics_opt_in: Optional[bool] = None, + stream_manager: Optional[StreamManager] = None, ) -> PlannerResult: """ Run the planner agent with tool-calling loop. @@ -91,12 +92,20 @@ def run( page: Current page URL (e.g. workflows/name/step-name) history: Conversation history stream: Whether to stream text via SSE events + stream_manager: Optional shared stream manager from the router, so + a handed-over request continues on the same stream Returns: PlannerResult with response, attachments, history, usage, meta """ 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 @@ -272,7 +281,7 @@ def _build_user_content(self, content: str, page: Optional[str]) -> str: matched_key, _ = find_job_in_yaml(self.current_yaml, step_name) step_name = matched_key or step_name if step_name: - user_content += f"\n\n(The user is currently viewing the step '{step_name}' — \"this step\" refers to it.)" + user_content += f"\n\n(The user is currently viewing the step '{step_name}'.)" else: user_content += f"\n\n(The user is currently viewing: {page})" @@ -488,19 +497,7 @@ def _execute_tool(self, tool_use_block, stream_manager, total_usage, tool_calls_ if single_key: job_keys.append(single_key) - if not self.current_yaml: - tool_result = "No workflow available to inspect." - elif not job_keys: - tool_result = "ERROR: No job keys provided." - else: - parts = [] - for job_key in job_keys: - _, job_data = find_job_in_yaml(self.current_yaml, job_key) - if job_data and job_data.get("body"): - parts.append(f"Job code for '{job_key}':\n\n{job_data['body']}") - else: - parts.append(f"No code found for job '{job_key}'.") - tool_result = "\n\n".join(parts) + tool_result = inspect_job_code(self.current_yaml, job_keys) tool_calls_meta.append({"tool": "inspect_job_code", "input": tool_use_block.input}) diff --git a/services/global_chat/prompts.yaml b/services/global_chat/prompts.yaml index 4da24ee0..10712118 100644 --- a/services/global_chat/prompts.yaml +++ b/services/global_chat/prompts.yaml @@ -108,8 +108,26 @@ prompts: Job code is stitched into the workflow YAML by job_key — the workflow must exist first. 1. Create/modify workflow structure FIRST (`call_workflow_agent`) - 2. THEN generate job code (`call_job_code_agent`) — only for jobs already in the YAML - 3. Set `job_key` to the exact key from the workflow structure + 2. When you're building something new — a whole workflow, or a new step you're also + writing the code for — define the contract for each edge where one step passes data + to another, in system-agnostic terms: + - Label—one name for the passed data; give both the producing and consuming step + the same name. + - Ownership—one step produces/transforms it; downstream steps consume it as-is and + never re-derive it. + + Put the identical contract line in both `call_job_code_agent` messages. Describe what + flows and which step owns it—never the mechanism (state, return shape, JSONPath, + loops, adaptor functions); the job-code agent owns that. + + e.g. "Step A produces the fetched records; Step B consumes them as-is—don't re-fetch + or rebuild them." + + This only applies to work you are creating. When editing existing steps, make only + the change the user asked for—don't restate contracts, re-derive the data flow, or + make other changes they didn't request. + 3. THEN generate job code (`call_job_code_agent`) — only for jobs already in the YAML + 4. Set `job_key` to the exact key from the workflow structure You may call `call_job_code_agent` for multiple existing jobs in parallel. Never call `call_job_code_agent` and `call_workflow_agent` in the same step. diff --git a/services/global_chat/router.py b/services/global_chat/router.py index db0bdcf3..f74e4191 100644 --- a/services/global_chat/router.py +++ b/services/global_chat/router.py @@ -17,11 +17,12 @@ sys.path.append(str(Path(__file__).parent.parent)) -from langfuse import observe +from langfuse import observe, get_client as get_langfuse_client from util import create_logger, ApolloError, sum_usage +from streaming_util import StreamManager from global_chat.config_loader import ConfigLoader from models import resolve_model -from global_chat.yaml_utils import get_step_name_from_page, find_job_in_yaml, stitch_job_code, workflow_has_job_code +from yaml_utils import get_step_name_from_page, get_page_view, find_job_in_yaml, stitch_job_code, workflow_has_job_code logger = create_logger(__name__) @@ -111,6 +112,10 @@ def route_and_execute( self._input_attachments = attachments or [] self._user = user self._metrics_opt_in = metrics_opt_in + # One stream manager shared by whichever agents serve this request, so + # a handed-over request continues the same stream instead of starting + # a second message lifecycle. + self._stream_manager = StreamManager(model=self.model, stream=stream) try: decision = self._make_routing_decision(content, workflow_yaml, page, history) @@ -121,8 +126,18 @@ def route_and_execute( logger.warning(f"Routing decision failed: {e}. Defaulting to planner for safety.") decision = RouterDecision(destination="planner", confidence=1) + # Direct routes are a fast path for clear-cut requests; when the router + # itself is unsure, take the path that can't be wrong. Costs nothing: + # the confidence comes back in the same routing call. + if decision.destination in ("workflow_agent", "job_code_agent") and decision.confidence < 3: + logger.warning( + f"Low router confidence ({decision.confidence}) for {decision.destination} — routing to planner instead" + ) + self._track_reroute({"low_confidence_reroute": decision.destination}) + decision = RouterDecision(destination="planner", confidence=decision.confidence) + if decision.destination == "workflow_agent": - result = self._route_to_workflow_chat(content, workflow_yaml, history, stream, decision.confidence) + result = self._route_to_workflow_chat(content, workflow_yaml, page, history, stream, decision.confidence) elif decision.destination == "job_code_agent": result = self._route_to_job_chat( content, workflow_yaml, page, history, stream, decision.confidence, decision.job_key @@ -229,7 +244,7 @@ def _format_attachments_for_content(self, content: str) -> str: return "\n".join(parts) def _route_to_workflow_chat( - self, content: str, workflow_yaml: Optional[str], history: List[Dict], stream: bool, confidence: int + self, content: str, workflow_yaml: Optional[str], page: Optional[str], history: List[Dict], stream: bool, confidence: int ) -> RouterResult: """Route directly to workflow_chat.""" from workflow_chat.workflow_chat import main as workflow_chat_main @@ -247,9 +262,17 @@ def _route_to_workflow_chat( "api_key": self.api_key, "meta": {"user": self._user} if self._user else None, "metrics_opt_in": self._metrics_opt_in, + "subagent": True, + "_stream_manager": self._stream_manager, } result = workflow_chat_main(payload) + + if result.get("handover"): + return self._handover_to_planner( + "workflow_agent", result, content, workflow_yaml, page, history, stream, confidence + ) + total_usage = sum_usage(self.routing_usage, result["usage"]) attachments = [] @@ -330,6 +353,22 @@ def _route_to_job_chat( job_context["adaptor"] = job_data["adaptor"] if job_data.get("name"): job_context["page_name"] = job_data["name"] + if matched_job_key: + # Tells job_chat's subagent prompt which step is focused/editable + job_context["job_key"] = matched_job_key + + # What the user actually has on screen, independent of which step we + # focus for editing: a specific step's code, or the workflow canvas. + # Only the router knows this (planner/prod calls omit it, so the prompt + # grounding line stays off). Fail safe: only claim a step the page name + # resolves to a real job — a mis-split name simply yields no line. + page_view, page_step = get_page_view(page) + if page_view == "step" and workflow_yaml: + _, viewed_job = find_job_in_yaml(workflow_yaml, page_step) + if viewed_job and viewed_job.get("name"): + job_context["viewing"] = viewed_job["name"] + elif page_view == "overview": + job_context["viewing"] = "canvas" clean_history = [{"role": t["role"], "content": t["content"]} for t in history] enriched_content = self._format_attachments_for_content(content) @@ -343,9 +382,18 @@ def _route_to_job_chat( "api_key": self.api_key, "meta": {"user": self._user} if self._user else None, "metrics_opt_in": self._metrics_opt_in, + "subagent": True, + "workflow_yaml": workflow_yaml, + "_stream_manager": self._stream_manager, } result = job_chat_main(payload) + + if result.get("handover"): + return self._handover_to_planner( + "job_code_agent", result, content, workflow_yaml, page, history, stream, confidence + ) + total_usage = sum_usage(self.routing_usage, result["usage"]) # Stitch suggested_code back into the workflow YAML. The full YAML is @@ -369,6 +417,46 @@ def _route_to_job_chat( meta={"agents": ["router", "job_code_agent"], "router_confidence": confidence}, ) + def _handover_to_planner( + self, + from_agent: str, + subagent_result: Dict, + content: str, + workflow_yaml: Optional[str], + page: Optional[str], + history: List[Dict], + stream: bool, + confidence: int, + ) -> RouterResult: + """Reroute a handed-over request to the planner. + + A direct-routed subagent signalled it cannot complete the request + (wrong route or missing capability). The planner never hands over, so + this retries at most once. The shared stream manager means the user + never sees the aborted attempt. + """ + reason = subagent_result["handover"] + logger.warning(f"{from_agent} handed over: {reason}. Rerouting to planner") + self._track_reroute({"handover_from": from_agent, "handover_reason": reason}) + + planner_result = self._route_to_planner(content, workflow_yaml, page, history, stream, confidence) + planner_result.usage = sum_usage(planner_result.usage, subagent_result.get("usage", {})) + return planner_result + + def _track_reroute(self, metadata: Dict) -> None: + """Record reroute diagnostics on the Langfuse trace (opt-in per request). + + Deliberately kept out of the response meta: the frontend does nothing + with these, they are for Langfuse analysis only. Server logs carry the + same information when tracking is off. + """ + if not self._metrics_opt_in: + return + try: + get_langfuse_client().update_current_span(metadata=metadata) + except Exception: + logger.warning("Failed to record reroute metadata in Langfuse") + def _route_to_planner( self, content: str, @@ -395,6 +483,7 @@ def _route_to_planner( stream=stream, user=self._user, metrics_opt_in=self._metrics_opt_in, + stream_manager=self._stream_manager, ) total_usage = sum_usage(self.routing_usage, planner_result.usage) diff --git a/services/global_chat/subagent_caller.py b/services/global_chat/subagent_caller.py index 19d90653..6c34ce1b 100644 --- a/services/global_chat/subagent_caller.py +++ b/services/global_chat/subagent_caller.py @@ -13,7 +13,7 @@ from langfuse import observe from util import create_logger, ApolloError -from global_chat.yaml_utils import find_job_in_yaml +from yaml_utils import find_job_in_yaml logger = create_logger(__name__) diff --git a/services/global_chat/tests/acceptance/job_code/test_canvas_code_request.md b/services/global_chat/tests/acceptance/job_code/test_canvas_code_request.md new file mode 100644 index 00000000..31f42445 --- /dev/null +++ b/services/global_chat/tests/acceptance/job_code/test_canvas_code_request.md @@ -0,0 +1,112 @@ +--- +id: global-chat.job-code.canvas-code-request +service: global_chat +judges: [general, openfn_code_quality] +--- + +# notes + +The user is on the workflow canvas (a 2-segment page URL, no step open) but asks +for a code change to a named step. The router should resolve this to +job_code_agent for the fetch-orders step, so job_chat is told the user is viewing +the canvas while fetch-orders is the step it can edit — the case where the +on-screen view and the editable step deliberately differ. + +Watch for two things. First, the edit must land on the fetch-orders step (not be +refused because "no step is open", and not bodged elsewhere). Second, the reply +must read like a normal answer: it must not surface internal mechanics (routing, +agents, subagents) or treat being on the canvas / not having a step open as a +limitation it narrates to the user. + +# quality_criteria + +- The fetch-orders step is updated to log a warning (e.g. a console.warn) when the API returns no orders — a guard that checks the fetched orders are empty. +- Only the fetch-orders step is changed; the other steps are left unchanged. +- The reply reads as a direct answer to the request and does NOT mention internal mechanics (routing, agents, subagents) or frame "being on the canvas" / "no step open" as a reason it cannot help. + +# settings + +## page + +workflows/orders-sync + +## workflow_yaml + +```yaml +name: orders-sync +jobs: + fetch-orders: + id: job-fetch-orders-id + name: Fetch Orders + adaptor: "@openfn/language-http@6.5.4" + body: | + get('/orders', { query: { since: $.lastRunAt } }); + fn(state => { + const orders = state.data.orders || []; + return { ...state, orders }; + }); + normalize-orders: + id: job-normalize-orders-id + name: Normalize Orders + adaptor: "@openfn/language-common@2.3.0" + body: | + fn(state => { + const orders = state.orders.map(o => ({ + id: o.id, + total: Number(o.total_price), + customerEmail: o.customer?.email, + placedAt: o.created_at + })); + return { ...state, orders }; + }); + notify-fulfillment: + id: job-notify-fulfillment-id + name: Notify Fulfillment + adaptor: "@openfn/language-http@6.5.4" + body: | + each( + $.orders, + post('https://fulfillment.example.org/queue', state => ({ + body: state.data + })) + ); +triggers: + cron: + id: trigger-cron-id + type: cron + cron_expression: "*/30 * * * *" + enabled: true +edges: + cron->fetch-orders: + id: edge-cron-fetch + source_trigger: cron + target_job: fetch-orders + condition_type: always + enabled: true + fetch-orders->normalize-orders: + id: edge-fetch-normalize + source_job: fetch-orders + target_job: normalize-orders + condition_type: on_job_success + enabled: true + normalize-orders->notify-fulfillment: + id: edge-normalize-notify + source_job: normalize-orders + target_job: notify-fulfillment + condition_type: on_job_success + enabled: true +``` + +## meta.session_id + +sess-job-code-canvas-code-request-0001 + +# turn + +## role + +user + +## content + +In the fetch-orders step, add a check that logs a warning if the API comes back with no orders. diff --git a/services/global_chat/tests/acceptance/job_code/test_question_reading_another_step.md b/services/global_chat/tests/acceptance/job_code/test_question_reading_another_step.md new file mode 100644 index 00000000..65c8be25 --- /dev/null +++ b/services/global_chat/tests/acceptance/job_code/test_question_reading_another_step.md @@ -0,0 +1,114 @@ +--- +id: global-chat.job-code.question-reading-another-step +service: global_chat +judges: [general, openfn_code_quality] +--- + +# notes + +The user is on the last step (notify-fulfillment) and asks what fields each order +has "at this point". The focused step only consumes `$.orders`; the shape those +orders actually have is defined by the UPSTREAM normalize-orders step, not by any +code visible in the focused step. To answer correctly the assistant has to read +the normalize-orders step and describe the fields it produces. + +This exercises the read-only path new to job_chat in subagent mode: it should +route to job_code_agent, use inspect_job_code to read the upstream step, and +answer — without escalating to the planner and without a code change. (The +planner could also field it; either way the answer must reflect the normalize +step's real output, not a generic guess.) The key failure mode to catch is the +model answering from thin air, or replying that it cannot see the data / the +other step. + +# quality_criteria + +- The response describes the normalized order shape produced upstream by the normalize-orders step: an id, a numeric total, a customerEmail, and a placedAt (timestamp). +- The answer is grounded in the actual upstream code, not a generic description of "an order", and it does NOT claim it cannot see the data or the other step. +- The response does NOT propose or apply a code change — the user asked a question. + +# settings + +## page + +workflows/orders-sync/notify-fulfillment + +## workflow_yaml + +```yaml +name: orders-sync +jobs: + fetch-orders: + id: job-fetch-orders-id + name: Fetch Orders + adaptor: "@openfn/language-http@6.5.4" + body: | + get('/orders', { query: { since: $.lastRunAt } }); + fn(state => { + const orders = state.data.orders || []; + return { ...state, orders }; + }); + normalize-orders: + id: job-normalize-orders-id + name: Normalize Orders + adaptor: "@openfn/language-common@2.3.0" + body: | + fn(state => { + const orders = state.orders.map(o => ({ + id: o.id, + total: Number(o.total_price), + customerEmail: o.customer?.email, + placedAt: o.created_at + })); + return { ...state, orders }; + }); + notify-fulfillment: + id: job-notify-fulfillment-id + name: Notify Fulfillment + adaptor: "@openfn/language-http@6.5.4" + body: | + each( + $.orders, + post('https://fulfillment.example.org/queue', state => ({ + body: state.data + })) + ); +triggers: + cron: + id: trigger-cron-id + type: cron + cron_expression: "*/30 * * * *" + enabled: true +edges: + cron->fetch-orders: + id: edge-cron-fetch + source_trigger: cron + target_job: fetch-orders + condition_type: always + enabled: true + fetch-orders->normalize-orders: + id: edge-fetch-normalize + source_job: fetch-orders + target_job: normalize-orders + condition_type: on_job_success + enabled: true + normalize-orders->notify-fulfillment: + id: edge-normalize-notify + source_job: normalize-orders + target_job: notify-fulfillment + condition_type: on_job_success + enabled: true +``` + +## meta.session_id + +sess-job-code-question-reading-another-step-0001 + +# turn + +## role + +user + +## content + +Before I post these to fulfillment, what fields does each order actually have at this point? diff --git a/services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md b/services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md new file mode 100644 index 00000000..a973260e --- /dev/null +++ b/services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md @@ -0,0 +1,92 @@ +--- +id: global-chat.rest-to-rest-sync-with-cron +service: global_chat +judges: [general, openfn_workflow_expert, openfn_code_quality] +--- + +# notes + +From-scratch scheduled REST-to-REST sync with fully specified job code. No existing YAML, no history. The user gives a precise spec: a daily cron trigger, an HTTP GET of a user list, a transform into a three-field shape (userId, title, body), and an HTTP POST of each transformed record. The planner should be invoked, call the workflow agent to produce the structure with a cron trigger, then call the job code agent to fill in the bodies. + +The key thing this test probes is data-flow coherence across steps: the transform step and the post step must agree on how the transformed records are passed between them. The steps should not read as if written in isolation — the downstream step must consume exactly what the upstream step produced, under the same name, without re-fetching or rebuilding it. + +The following workflow is a NON-BINDING reference showing one acceptable shape. Do not require the candidate to match it (adaptor versions, job names, whether GET and transform are one step or two, and the exact state key may all differ). Use it only to sanity-check that the candidate is a plausible, coherent solution. + +```yaml +name: " Daily REST Endpoint Sync (Manual)" +jobs: + Fetch-and-transform-users: + name: Fetch and transform users + adaptor: "@openfn/language-http@latest" + body: >- + + + get('https://jsonplaceholder.typicode.com/users'); + + + fn((state) => { + const users = state.data || []; + const records = users.map((user) => ({ + userId: user.id, + title: user.name, + body: `Email: ${user.email} | Company: ${user.company?.name ?? 'N/A'}`, + })); + console.log(`Transformed ${records.length} users`); + return { ...state, records }; + }); + Post-records-to-target: + name: Post records to target + adaptor: "@openfn/language-http@7.3.2" + body: > + each( + '$.records[*]', + post('https://jsonplaceholder.typicode.com/posts', (state) => state.data) + ); + + + fn((state) => { + console.log(`Posted ${state.records?.length ?? 0} records`); + return state; + }); +triggers: + cron: + type: cron + enabled: false + cron_expression: 0 0 * * * + cron_cursor_job: null +edges: + cron->Fetch-and-transform-users: + condition_type: always + enabled: true + target_job: Fetch-and-transform-users + source_trigger: cron + Fetch-and-transform-users->Post-records-to-target: + condition_type: on_job_success + enabled: true + target_job: Post-records-to-target + source_job: Fetch-and-transform-users +``` + +# quality_criteria + +- The workflow uses a cron trigger scheduled to run once a day (e.g. a `0 0 * * *` daily expression), not a webhook or a different frequency. +- A step fetches the user list from the source endpoint (`https://jsonplaceholder.typicode.com/users`) using an HTTP get. +- A transform maps each user into an object with exactly the three requested fields: `userId` (the user's id), `title` (the user's name), and `body` (a string combining the user's email and company name). +- A step POSTs each transformed record to the target endpoint (`https://jsonplaceholder.typicode.com/posts`) using an HTTP post. +- Data-flow coherence: the posting step consumes the exact data the transform step produced, referencing it under the same state key/name that the transform step wrote to. There is no key mismatch between the producing and consuming steps. +- The posting step does not re-fetch the users or rebuild the transformed objects itself — it consumes the upstream output as-is rather than duplicating the transform. +- The solution stays simple as requested: no branching, filtering, deduplication, or auth logic beyond what the user asked for. + +# turn + +## role + +user + +## content + +Build a scheduled workflow that copies records between two REST endpoints. +Trigger: cron, once a day. +Steps: GET the list of users from https://jsonplaceholder.typicode.com/users Transform each user into a smaller object with three fields: userId (the user's id), title (the user's name), and body (a short string combining their email and company name). POST each transformed record to https://jsonplaceholder.typicode.com/posts + +No authentication is required for this API. Keep it simple: no branching or deduplication. diff --git a/services/global_chat/tests/integration/__init__.py b/services/global_chat/tests/integration/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/services/global_chat/tests/integration/test_handover.py b/services/global_chat/tests/integration/test_handover.py new file mode 100644 index 00000000..f17521d2 --- /dev/null +++ b/services/global_chat/tests/integration/test_handover.py @@ -0,0 +1,136 @@ +"""Integration tests for subagent handover: curated misroutes sent directly to +job_chat and workflow_chat, as if the router had picked the wrong destination. + +These hit the live Anthropic API (manual/nightly, costs tokens). Each service +gets two scenarios: an obvious misroute, and an oblique one where the request +never names the other step — the model has to work out that what's being asked +lies beyond what it can see or edit. +""" + +import pytest +from dotenv import load_dotenv + +load_dotenv() + +from job_chat.job_chat import main as job_chat_main # noqa: E402 +from workflow_chat.workflow_chat import main as workflow_chat_main # noqa: E402 + +pytestmark = pytest.mark.integration + +WORKFLOW_YAML = """\ +name: patient-sync +jobs: + fetch-patients: + id: 5f2f36e7-3f42-4b0a-9c11-9b3f5a3d1a01 + name: Fetch Patients + adaptor: "@openfn/language-http@latest" + body: | + get('/patients'); + notify-admin: + id: 8a1c22d0-6e4f-49d3-b6a2-4bfeb1f0c902 + name: Notify Admin + adaptor: "@openfn/language-http@latest" + body: | + fn(state => { + if (!state.data || state.data.length === 0) { + console.warn('SYNC-WARN-042: no patient recrods found'); + } + return state; + }); +triggers: + webhook: + id: 2d7f80b3-1c55-47a9-8e2f-6a90d24c7a03 + type: webhook +edges: + webhook->fetch-patients: + id: c4e9a1f6-0d82-4c37-9b54-7e315f68bd04 + source_trigger: webhook + target_job: fetch-patients + condition_type: always +""" + + +def job_chat_payload(content: str) -> dict: + return { + "content": content, + "suggest_code": True, + "subagent": True, + "workflow_yaml": WORKFLOW_YAML, + "context": { + "expression": "get('/patients');", + "adaptor": "@openfn/language-http@latest", + "page_name": "Fetch Patients", + "job_key": "fetch-patients", + }, + } + + +def workflow_chat_payload(content: str) -> dict: + return { + "content": content, + "subagent": True, + "existing_yaml": WORKFLOW_YAML, + } + + +def test_job_chat_hands_over_structural_request() -> None: + """A workflow-structure request misrouted to job_chat must hand over, + with no user-visible reply text and no code attached.""" + result = job_chat_main(job_chat_payload( + "Add a new step after this one that sends the patients to Salesforce, and connect it up", + )) + + assert result.get("handover") + assert result["response"] == "" + assert result.get("suggested_code") is None + + +def test_job_chat_hands_over_oblique_other_step_edit() -> None: + """An edit whose target code lives in another step, described by what it + does rather than by name, must hand over — not be bodged into the focused + step or met with 'I can't see that code'.""" + result = job_chat_main(job_chat_payload( + "The warning we log when no patients are found should include the run date too, can you update it?", + )) + + assert result.get("handover") + assert result["response"] == "" + assert result.get("suggested_code") is None + + +def test_workflow_chat_hands_over_code_request() -> None: + """A job-code request misrouted to workflow_chat must hand over, + with no user-visible reply text and no YAML.""" + result = workflow_chat_main(workflow_chat_payload( + "Why does the code in my fetch-patients step return an empty array? Can you fix it?", + )) + + assert result.get("handover") + assert result["response"] == "" + assert not result.get("response_yaml") + + +def test_job_chat_hands_over_when_focused_step_is_wrong() -> None: + """Right subagent, wrong step: the user says "this step" but the code they + describe lives in a different step than the one the router focused. The + model must not claim it can't see the warning, and must not bodge a new + warning into the wrong step.""" + result = job_chat_main(job_chat_payload( + "Fix the typo in the warning message this step logs when there are no patients", + )) + + assert result.get("handover") + assert result["response"] == "" + assert result.get("suggested_code") is None + + +def test_workflow_chat_hands_over_oblique_code_change() -> None: + """A code-level change described without naming any step must hand over — + workflow_chat sees only redacted job bodies and cannot make it.""" + result = workflow_chat_main(workflow_chat_payload( + "Can you change the wording of the warning we log when no patients come back?", + )) + + assert result.get("handover") + assert result["response"] == "" + assert not result.get("response_yaml") diff --git a/services/global_chat/tests/unit/test_router.py b/services/global_chat/tests/unit/test_router.py index 1a3ada34..8f97c499 100644 --- a/services/global_chat/tests/unit/test_router.py +++ b/services/global_chat/tests/unit/test_router.py @@ -2,8 +2,8 @@ from unittest.mock import patch -from global_chat.router import RouterAgent -from global_chat.yaml_utils import workflow_has_job_code +from global_chat.router import RouterAgent, RouterDecision, RouterResult +from yaml_utils import workflow_has_job_code EMPTY_YAML = """\ name: wf @@ -33,6 +33,8 @@ def make_router() -> RouterAgent: router._input_attachments = [] router._user = None router._metrics_opt_in = None + router._stream_manager = None + router.model = "claude-test" return router @@ -87,3 +89,100 @@ def test_routing_message_tags_empty_workflow() -> None: router = make_router() msg = router._build_routing_message("what does this do", EMPTY_YAML, None, []) assert "[All step bodies are empty/placeholder]" in msg + + +def test_job_route_sends_subagent_payload() -> None: + router = make_router() + + with patch("job_chat.job_chat.main", return_value=job_chat_result(None)) as mock_main: + router._route_to_job_chat( + "explain this", WORKFLOW_YAML, "workflows/wf/fetch-patients", [], False, 5, + ) + + payload = mock_main.call_args[0][0] + assert payload["subagent"] is True + assert payload["workflow_yaml"] == WORKFLOW_YAML + assert payload["context"]["job_key"] == "fetch-patients" + + +def make_planner_result() -> RouterResult: + return RouterResult( + response="planner answer", + response_segments=[{"type": "text", "content": "planner answer"}], + attachments=[], + history=[], + usage={"input_tokens": 10}, + meta={"agents": ["router", "planner"]}, + ) + + +def test_job_route_handover_reroutes_to_planner() -> None: + router = make_router() + handed_over = {"response": "", "handover": "needs structure changes", "history": [], "usage": {"input_tokens": 7}} + + with patch("job_chat.job_chat.main", return_value=handed_over), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()) as planner_mock: + result = router._route_to_job_chat( + "add a step", WORKFLOW_YAML, "workflows/wf/fetch-patients", [], False, 5, + ) + + planner_mock.assert_called_once() + assert result.response == "planner answer" + # Reroute diagnostics stay out of the response meta (Langfuse-only) + assert "handover_from" not in result.meta + # Usage from the aborted job_chat call is kept on top of the planner's + assert result.usage["input_tokens"] == 17 + + +def test_workflow_route_handover_reroutes_to_planner() -> None: + router = make_router() + handed_over = { + "response": "", "response_yaml": None, "handover": "asks about job code", + "history": [], "usage": {"input_tokens": 3}, + } + + with patch("workflow_chat.workflow_chat.main", return_value=handed_over), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()) as planner_mock: + result = router._route_to_workflow_chat( + "what does this code do", WORKFLOW_YAML, "workflows/wf", [], False, 4, + ) + + planner_mock.assert_called_once() + assert "handover_from" not in result.meta + assert result.usage["input_tokens"] == 13 + + +def test_low_confidence_direct_route_goes_to_planner() -> None: + router = make_router() + decision = RouterDecision(destination="job_code_agent", confidence=2, job_key="fetch-patients") + + with patch.object(RouterAgent, "_make_routing_decision", return_value=decision), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()) as planner_mock, \ + patch.object(RouterAgent, "_route_to_job_chat") as job_mock: + result = router.route_and_execute("edit this", WORKFLOW_YAML, None, [], False) + + planner_mock.assert_called_once() + job_mock.assert_not_called() + assert result.response == "planner answer" + + +def test_confident_direct_route_is_not_gated() -> None: + router = make_router() + decision = RouterDecision(destination="job_code_agent", confidence=3, job_key="fetch-patients") + job_result = RouterResult( + response="job answer", + response_segments=[{"type": "text", "content": "job answer"}], + attachments=[], + history=[], + usage={}, + meta={}, + ) + + with patch.object(RouterAgent, "_make_routing_decision", return_value=decision), \ + patch.object(RouterAgent, "_route_to_planner") as planner_mock, \ + patch.object(RouterAgent, "_route_to_job_chat", return_value=job_result) as job_mock: + result = router.route_and_execute("edit this", WORKFLOW_YAML, None, [], False) + + job_mock.assert_called_once() + planner_mock.assert_not_called() + assert result.response == "job answer" diff --git a/services/global_chat/tests/unit/test_yaml_utils.py b/services/global_chat/tests/unit/test_yaml_utils.py new file mode 100644 index 00000000..024965f4 --- /dev/null +++ b/services/global_chat/tests/unit/test_yaml_utils.py @@ -0,0 +1,68 @@ +"""Unit tests for the shared inspect_job_code tool executor and redaction.""" + +from yaml_utils import inspect_job_code, redact_job_bodies + +WORKFLOW_YAML = """\ +name: wf +jobs: + fetch-patients: + name: Fetch Patients + body: get('/patients'); + send-data: + name: Send Data + body: post('/data', $.data); +""" + +WORKFLOW_YAML_WITH_IDS = """\ +name: wf +jobs: + fetch-patients: + id: 5f2f36e7-3f42-4b0a-9c11-9b3f5a3d1a01 + name: Fetch Patients + adaptor: "@openfn/language-http@latest" + body: get('/patients'); +triggers: + webhook: + id: 2d7f80b3-1c55-47a9-8e2f-6a90d24c7a03 + type: webhook +edges: + webhook->fetch-patients: + id: c4e9a1f6-0d82-4c37-9b54-7e315f68bd04 + source_trigger: webhook + target_job: fetch-patients + condition_type: always +""" + + +def test_redact_strips_bodies_and_ids_keeps_structure() -> None: + redacted = redact_job_bodies(WORKFLOW_YAML_WITH_IDS) + + assert "get('/patients');" not in redacted + assert "# [use inspect_job_code to view]" in redacted + assert "id:" not in redacted + # Structure the model needs stays intact + assert "Fetch Patients" in redacted + assert "@openfn/language-http@latest" in redacted + assert "webhook->fetch-patients" in redacted + assert "condition_type: always" in redacted + + +def test_inspect_returns_requested_bodies() -> None: + result = inspect_job_code(WORKFLOW_YAML, ["fetch-patients", "send-data"]) + assert "get('/patients');" in result + assert "post('/data', $.data);" in result + + +def test_inspect_matches_fuzzy_names() -> None: + result = inspect_job_code(WORKFLOW_YAML, ["Fetch Patients"]) + assert "get('/patients');" in result + + +def test_inspect_reports_missing_job() -> None: + result = inspect_job_code(WORKFLOW_YAML, ["nonexistent"]) + assert "No code found for job 'nonexistent'." in result + + +def test_inspect_handles_missing_yaml_and_keys() -> None: + assert inspect_job_code(None, ["a"]) == "No workflow available to inspect." + assert inspect_job_code(WORKFLOW_YAML, []) == "ERROR: No job keys provided." diff --git a/services/global_chat/tools/tool_definitions.py b/services/global_chat/tools/tool_definitions.py index 6f94fa7b..6322e20e 100644 --- a/services/global_chat/tools/tool_definitions.py +++ b/services/global_chat/tools/tool_definitions.py @@ -73,24 +73,9 @@ "cache_control": {"type": "ephemeral"} } -# Tool 4: Inspect job code -INSPECT_JOB_CODE_TOOL = { - "name": "inspect_job_code", - "description": """Read the current code body of one or more jobs in the workflow (read-only). - -Use this to inspect existing step code before editing — e.g. to find which steps a change applies to before editing only those, or to base one step on another. Pass all the job keys you need in a single call rather than calling once per job.""", - "input_schema": { - "type": "object", - "properties": { - "job_keys": { - "type": "array", - "items": {"type": "string"}, - "description": "The job keys to inspect (e.g. ['fetch-patients', 'load-dhis2'])" - } - }, - "required": ["job_keys"] - } -} +# Tool 4: Inspect job code — shared with job_chat's subagent mode so both +# agents explore the workflow with the exact same tool +from yaml_utils import INSPECT_JOB_CODE_TOOL # noqa: E402 # Export all tool definitions TOOL_DEFINITIONS = [ diff --git a/services/global_chat/yaml_utils.py b/services/global_chat/yaml_utils.py deleted file mode 100644 index 119bbcf0..00000000 --- a/services/global_chat/yaml_utils.py +++ /dev/null @@ -1,123 +0,0 @@ -""" -Shared YAML utility functions for working with workflow YAML strings. - -Used by router and subagent caller for job extraction and code stitching. -""" -import re -import yaml -from typing import Dict, Optional, Tuple - - -def get_step_name_from_page(page: Optional[str]) -> Optional[str]: - """ - Extract step name from page URL. - - Examples: - workflows/my-workflow/fetch-patients -> "fetch-patients" - workflows/my-workflow -> None - workflows/my-workflow/settings -> None - """ - if not page: - return None - - parts = page.strip("/").split("/") - if len(parts) == 3 and parts[0] == "workflows" and parts[2] != "settings": - return parts[2] - - return None - - -def normalize_name(name: str) -> str: - """Normalize a name for fuzzy matching: lowercase, non-alphanumeric chars become hyphens.""" - return re.sub(r'[^a-z0-9]', '-', name.lower()).strip('-') - - -def find_job_in_yaml(yaml_str: str, step_name: str) -> Tuple[Optional[str], Optional[Dict]]: - """ - Find a job in the workflow YAML by step name. - - Tries direct key match first, then normalized name comparison against - both the job key and the job's name field. - - Returns: - (job_key, job_data) or (None, None) if not found or on parse error - """ - try: - yaml_data = yaml.safe_load(yaml_str) - except Exception: - return None, None - - if not yaml_data or "jobs" not in yaml_data: - return None, None - - jobs = yaml_data["jobs"] - - # Direct key match - if step_name in jobs: - return step_name, jobs[step_name] - - # Normalized match: compare against job key and name field - normalized_step = normalize_name(step_name) - for job_key, job_data in jobs.items(): - if normalize_name(job_key) == normalized_step: - return job_key, job_data - job_name = job_data.get("name", "") - if normalize_name(job_name) == normalized_step: - return job_key, job_data - - return None, None - - -EMPTY_JOB_BODY = "// Add operations here" - - -def workflow_has_job_code(yaml_str: Optional[str]) -> bool: - """Return True if any job has a non-empty, non-placeholder body. - - The canonical empty-job marker is ``// Add operations here`` (see - workflow_chat); a blank body or that marker means "no code yet". Used to - decide whether a "what does this do" question needs the planner (to read the - real code) or can take the faster workflow_agent path (structure only). - """ - try: - yaml_data = yaml.safe_load(yaml_str) - except Exception: - return False - if not yaml_data or "jobs" not in yaml_data: - return False - for job_data in yaml_data["jobs"].values(): - body = (job_data or {}).get("body") - if isinstance(body, str) and body.strip() and body.strip() != EMPTY_JOB_BODY: - return True - return False - - -def redact_job_bodies(yaml_str: str) -> str: - """Return workflow YAML with job bodies replaced by a placeholder.""" - try: - yaml_data = yaml.safe_load(yaml_str) - if yaml_data and "jobs" in yaml_data: - for job_data in yaml_data["jobs"].values(): - if "body" in job_data: - job_data["body"] = "# [use inspect_job_code to view]" - return yaml.dump(yaml_data, sort_keys=False) - except Exception: - pass - return yaml_str - - -def stitch_job_code(yaml_str: str, job_key: str, new_code: str) -> str: - """ - Replace a job's body in the workflow YAML with new code. - - Returns the original YAML string unchanged if parsing or stitching fails. - """ - try: - yaml_data = yaml.safe_load(yaml_str) - if yaml_data and "jobs" in yaml_data and job_key in yaml_data["jobs"]: - yaml_data["jobs"][job_key]["body"] = new_code - return yaml.dump(yaml_data, sort_keys=False) - except Exception: - pass - - return yaml_str diff --git a/services/job_chat/job_chat.py b/services/job_chat/job_chat.py index 8a52f4fd..368f8573 100644 --- a/services/job_chat/job_chat.py +++ b/services/job_chat/job_chat.py @@ -20,6 +20,7 @@ from langfuse import observe, propagate_attributes, get_client as get_langfuse_client from langfuse_util import should_track, build_tags, build_generation_diff from util import ApolloError, create_logger, AdaptorSpecifier, add_page_prefix, APOLLO_VERSION +from yaml_utils import INSPECT_JOB_CODE_TOOL, inspect_job_code from .prompt import build_prompt, build_error_correction_prompt from .old_prompt import build_old_prompt from streaming_util import ( @@ -82,6 +83,49 @@ }, } +# Subagent mode only (job_chat called from global_chat): escalation disguised +# as a capability. Calling it hands the request back to the caller, which +# reroutes to the planner — so if the model narrates before calling, the +# narration ("I'll take a look at your workflow") matches what happens next. +_EDIT_WORKFLOW_TOOL = { + "name": "edit_workflow", + "description": ( + "Open the full workflow to work on anything beyond this step's code: " + "workflow structure (add/remove/rename steps, triggers, edges, adaptors) " + "or code changes in other steps. Call this as your VERY FIRST action. " + "To merely READ another step's code, use inspect_job_code instead." + ), + "input_schema": { + "type": "object", + "properties": { + "goal": { + "type": "string", + "description": "One sentence: what needs to be done", + } + }, + "required": ["goal"], + "additionalProperties": False, + }, +} + +# The planner's inspect tool (same name, schema, and executor via yaml_utils), +# with the description rewritten for job_chat: unlike the planner, job_chat can +# only edit the focused step, so reading must never look like a way to act. +_INSPECT_JOB_CODE_TOOL = { + **INSPECT_JOB_CODE_TOOL, + "description": ( + "Read the current code of one or more other steps in the workflow. " + "Use it when seeing another step's code helps you answer a question or " + "edit the focused step — e.g. to match its pattern, or to see the state " + "shape it produces. To change another step's code, call edit_workflow " + "instead. Pass all the job keys you need in a single call." + ), +} + +# Max API rounds in one generate() call: enough for a couple of +# inspect_job_code round-trips plus the final answer. +_MAX_TOOL_ROUNDS = 4 + # Helper function for page navigation def extract_page_prefix_from_last_turn(history: List[Dict[str, str]]) -> Optional[str]: @@ -116,6 +160,11 @@ class Payload: download_adaptor_docs: Optional[bool] = True refresh_rag: Optional[bool] = False metrics_opt_in: Optional[bool] = None + # Subagent mode: set only when called from global_chat, never by direct + # production callers. workflow_yaml additionally enables the + # inspect_job_code tool. + workflow_yaml: Optional[str] = None + subagent: Optional[bool] = False @classmethod def from_dict(cls, data: Dict[str, Any]) -> "Payload": @@ -135,6 +184,8 @@ def from_dict(cls, data: Dict[str, Any]) -> "Payload": download_adaptor_docs=data.get("download_adaptor_docs", True), refresh_rag=data.get("refresh_rag", False), metrics_opt_in=data.get("metrics_opt_in"), + workflow_yaml=data.get("workflow_yaml"), + subagent=data.get("subagent", False), ) @@ -153,6 +204,8 @@ class ChatResponse: usage: Dict[str, Any] rag: Dict[str, Any] diff: Optional[Dict[str, Any]] = None + # Subagent mode only: reason the request was handed back to the caller + handover: Optional[str] = None class AnthropicClient: def __init__(self, config: Optional[ChatConfig] = None): @@ -186,17 +239,25 @@ def generate( stream: Optional[bool] = False, download_adaptor_docs: Optional[bool] = True, refresh_rag: Optional[bool] = False, - current_page: Optional[dict] = None + current_page: Optional[dict] = None, + workflow_yaml: Optional[str] = None, + subagent: Optional[bool] = False, + stream_manager: Optional[StreamManager] = None, ) -> ChatResponse: """ Generate a response using the Claude API with optional streaming. + + In subagent mode (called from global_chat) the model can also read + other steps' code via inspect_job_code and hand the request back via + handover. A stream_manager may be injected by the caller so a handed- + over request continues on the same stream. """ sentry_sdk.set_tag("prompt_type", "code_suggestions" if suggest_code else "no_code_suggestions") with sentry_sdk.start_transaction(name="chat_generation") as transaction: history = history.copy() if history else [] - stream_manager = StreamManager(model=self.config.model, stream=stream) + stream_manager = stream_manager or StreamManager(model=self.config.model, stream=stream) if context and context.get("expression"): stream_manager.send_thinking(STATUS_REVIEWING_CODE) else: @@ -212,7 +273,9 @@ def generate( api_key=self.api_key, stream_manager=stream_manager, download_adaptor_docs=download_adaptor_docs, - refresh_rag=refresh_rag + refresh_rag=refresh_rag, + workflow_yaml=workflow_yaml, + subagent=subagent ) else: @@ -230,74 +293,148 @@ def generate( # tool. tool_choice stays "auto": the model answers in text # and only calls the tool when it actually wants to change the job. output_config = {"effort": "medium"} - tool_kwargs = ( - {"tools": [_EDIT_TOOL], "tool_choice": {"type": "auto"}} - if suggest_code else {} - ) + tools = [] + if suggest_code: + tools.append(_EDIT_TOOL) + if subagent: + tools.append(_EDIT_WORKFLOW_TOOL) + if workflow_yaml: + tools.append(_INSPECT_JOB_CODE_TOOL) + tool_kwargs = {"tools": tools, "tool_choice": {"type": "auto"}} if tools else {} + + # Without the subagent tools this loop runs exactly once: edit_job + # is terminal (its input IS the output), so only inspect_job_code + # triggers another round and only handover exits early. + messages = prompt + handover_reason = None + text_parts = [] + usage_events = [] with sentry_sdk.start_span(description="anthropic_api_call"): - if stream: - logger.info("Making streaming API call") - text_started = False - sent_length = 0 - accumulated_response = "" - self._stream_applied = False - self._stream_suggested_code = None - self._stream_diff = None - - original_code = context.get("expression") if context and isinstance(context, dict) else None - - stream_kwargs = dict( - max_tokens=self.config.max_tokens, - messages=prompt, - model=self.config.model, - system=system_message, - thinking={"type": "adaptive"}, - output_config=output_config, - **tool_kwargs - ) + for round_index in range(_MAX_TOOL_ROUNDS): + if stream: + logger.info("Making streaming API call") + text_started = False + sent_length = 0 + accumulated_response = "" + self._stream_applied = False + self._stream_suggested_code = None + self._stream_diff = None + + original_code = context.get("expression") if context and isinstance(context, dict) else None + + stream_kwargs = dict( + max_tokens=self.config.max_tokens, + messages=messages, + model=self.config.model, + system=system_message, + thinking={"type": "adaptive"}, + output_config=output_config, + **tool_kwargs + ) - with self.client.messages.stream(**stream_kwargs) as stream_obj: - for event in stream_obj: - if event.type == "message_start": - stream_manager.send_thinking(STATUS_WORKING) - # The edit_job tool block starts after the text ends; its - # input (the code) streams silently, so show a status here. - elif event.type == "content_block_start" and getattr(getattr(event, "content_block", None), "type", None) == "tool_use": - stream_manager.send_thinking(STATUS_WRITING_CODE) - accumulated_response, text_started, sent_length = self.process_stream_event( - event, - accumulated_response, - suggest_code, - text_started, - sent_length, - stream_manager, - original_code, - content - ) - message = stream_obj.get_final_message() + with self.client.messages.stream(**stream_kwargs) as stream_obj: + for event in stream_obj: + if event.type == "message_start" and round_index == 0: + stream_manager.send_thinking(STATUS_WORKING) + # The edit_job tool block starts after the text ends; its + # input (the code) streams silently, so show a status here. + elif event.type == "content_block_start" and getattr(getattr(event, "content_block", None), "type", None) == "tool_use" and getattr(getattr(event, "content_block", None), "name", None) == "edit_job": + stream_manager.send_thinking(STATUS_WRITING_CODE) + accumulated_response, text_started, sent_length = self.process_stream_event( + event, + accumulated_response, + suggest_code, + text_started, + sent_length, + stream_manager, + original_code, + content + ) + message = stream_obj.get_final_message() + + # Flush any remaining buffered text, stripping JSON closing chars + if suggest_code and text_started: + if sent_length < len(accumulated_response): + remaining = accumulated_response[sent_length:] + remaining = re.sub(r'"\s*}\s*$', '', remaining) + if remaining: + stream_manager.send_text(self._unescape_json_string(remaining)) + + else: + logger.info("Making non-streaming API call") + create_kwargs = dict( + max_tokens=self.config.max_tokens, messages=messages, model=self.config.model, system=system_message, + thinking={"type": "adaptive"}, + output_config=output_config, + # Per-request timeout (same values as the SDK default): + # required for non-streaming calls with max_tokens > ~21k, + # which the SDK otherwise rejects. + timeout=httpx.Timeout(600.0, connect=5.0), + **tool_kwargs + ) + message = self.client.messages.create(**create_kwargs) - # Flush any remaining buffered text, stripping JSON closing chars - if suggest_code and text_started: - if sent_length < len(accumulated_response): - remaining = accumulated_response[sent_length:] - remaining = re.sub(r'"\s*}\s*$', '', remaining) - if remaining: - stream_manager.send_text(self._unescape_json_string(remaining)) + if hasattr(message, "usage"): + usage_events.append(message.usage.model_dump()) - else: - logger.info("Making non-streaming API call") - create_kwargs = dict( - max_tokens=self.config.max_tokens, messages=prompt, model=self.config.model, system=system_message, - thinking={"type": "adaptive"}, - output_config=output_config, - # Per-request timeout (same values as the SDK default): - # required for non-streaming calls with max_tokens > ~21k, - # which the SDK otherwise rejects. - timeout=httpx.Timeout(600.0, connect=5.0), - **tool_kwargs - ) - message = self.client.messages.create(**create_kwargs) + for content_block in message.content: + if getattr(content_block, "type", None) == "text" and content_block.text: + text_parts.append(content_block.text) + + tool_uses = [b for b in message.content if getattr(b, "type", None) == "tool_use"] + if tool_uses: + logger.info( + "job_chat round %d: model called %s", + round_index, ", ".join(b.name for b in tool_uses), + ) + + handover_block = next((b for b in tool_uses if b.name == "edit_workflow"), None) + if handover_block: + handover_reason = (handover_block.input or {}).get("goal") or "handover requested" + break + + inspect_blocks = [b for b in tool_uses if b.name == "inspect_job_code"] + if not inspect_blocks or round_index == _MAX_TOOL_ROUNDS - 1: + break + + # Answer the inspect calls and let the model continue. An + # edit_job call made in the same round is deferred: the + # model must re-issue it with its final answer. + stream_manager.send_thinking(STATUS_REVIEWING_CODE) + tool_results = [] + for block in tool_uses: + if block.name == "inspect_job_code": + job_keys = (block.input or {}).get("job_keys") or [] + logger.info("job_chat inspect_job_code: reading %s", job_keys) + result_text = inspect_job_code(workflow_yaml, job_keys) + else: + result_text = ( + "Not applied. Finish inspecting, then write your final reply " + "and call edit_job again with the complete edits." + ) + tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": result_text}) + + messages = messages + [ + {"role": "assistant", "content": message.content}, + {"role": "user", "content": tool_results}, + ] + + if handover_reason: + logger.info(f"job_chat handing over: {handover_reason}") + # Deliberately do NOT end the stream: the caller reroutes the + # request and the next agent continues on the same stream. + return ChatResponse( + response="", + suggested_code=None, + history=history, + usage=self.sum_usage( + *usage_events, + *[usage_data for usage_key, usage_data in retrieved_knowledge.get("usage", {}).items()] + ), + rag=retrieved_knowledge, + handover=handover_reason, + ) if hasattr(message, "usage"): if message.usage.cache_creation_input_tokens: @@ -306,15 +443,13 @@ def generate( logger.info(f"Cache read: {message.usage.cache_read_input_tokens} tokens") # The model answers in normal text; it calls the `edit_job` tool only - # when it wants to change the user's job. So text = the reply, and the - # tool's parsed input carries the code edits (no JSON-in-text parsing). - text_parts = [] + # when it wants to change the user's job. So text = the reply + # (accumulated across tool rounds above), and the tool's parsed + # input carries the code edits (no JSON-in-text parsing). tool_code_edits = None for content_block in message.content: if getattr(content_block, "type", None) == "tool_use" and getattr(content_block, "name", None) == "edit_job": tool_code_edits = (content_block.input or {}).get("code_edits") or [] - elif getattr(content_block, "type", None) == "text": - text_parts.append(content_block.text) text_response = "\n\n".join(text_parts).strip() suggested_code = None @@ -355,7 +490,7 @@ def generate( ] usage = self.sum_usage( - message.usage.model_dump() if hasattr(message, "usage") else {}, + *usage_events, *[usage_data for usage_key, usage_data in retrieved_knowledge.get("usage", {}).items()] ) @@ -608,7 +743,7 @@ def main(data_dict: dict) -> dict: """ try: sentry_sdk.set_context("request_data", { - k: v for k, v in data_dict.items() if k != "api_key" + k: v for k, v in data_dict.items() if k not in ("api_key", "_stream_manager") }) data = Payload.from_dict(data_dict) @@ -667,7 +802,12 @@ def main(data_dict: dict) -> dict: stream=data.stream, download_adaptor_docs=data.download_adaptor_docs, refresh_rag=should_refresh_rag, - current_page=current_page + current_page=current_page, + workflow_yaml=data.workflow_yaml, + subagent=data.subagent, + # In-process callers (global_chat) may inject a shared stream + # manager so a handed-over request continues the same stream + stream_manager=data_dict.get("_stream_manager"), ) # Tag the trace when code was generated, so we can filter for it. @@ -675,6 +815,12 @@ def main(data_dict: dict) -> dict: with propagate_attributes(tags=["has_code_attachment"]): pass + # Tag the trace when the request was handed back for rerouting to + # the planner, so we can filter for handovers. + if tracking and result.handover: + with propagate_attributes(tags=["handover"]): + pass + if tracking: diff_meta = build_generation_diff( original=data.context.get("expression"), @@ -694,6 +840,9 @@ def main(data_dict: dict) -> dict: if result.diff: response_dict["diff"] = result.diff + if result.handover: + response_dict["handover"] = result.handover + return response_dict except ApolloError: diff --git a/services/job_chat/prompt.py b/services/job_chat/prompt.py index 1dd27dfc..fb1afc11 100644 --- a/services/job_chat/prompt.py +++ b/services/job_chat/prompt.py @@ -4,12 +4,13 @@ import sentry_sdk from langfuse import observe from util import create_logger, ApolloError, AdaptorSpecifier, get_db_connection +from yaml_utils import redact_job_bodies, normalize_name from .retrieve_docs import retrieve_knowledge from search_adaptor_docs.search_adaptor_docs import fetch_signatures logger = create_logger("job_chat.prompt") -system_role = """ +_role_before_scope = """ You are a software engineer helping a non-expert user write a job for our platform. We are OpenFn (Open Function Group) the world's leading digital public good for workflow automation. @@ -32,10 +33,17 @@ Your chat panel is embedded in a web based IDE, which lets users build a Workflow with a number of steps (or jobs). There is a code editor next to you, which users can copy and paste code into. Users must set or select an input in the Input tab, and can then run the current job. +""" +# Production only. In global chat's subagent mode, structure requests are +# handled via edit_workflow, so this scope restriction is omitted entirely +# rather than contradicted. +production_scope_instructions = """ You ONLY help with job code. Do NOT help with overall workflow structure. If the user wants to add/remove/edit workflow steps, tell them to navigate to the workflow overview. +""" +_role_after_scope = """ Users can Flag any answers that are not helpful, which will help us build a better prompt for you. @@ -60,6 +68,9 @@ """ +system_role = _role_before_scope + production_scope_instructions + _role_after_scope +subagent_system_role = _role_before_scope + _role_after_scope + job_writing_summary = """ When writing jobs, users will use their own credentials to access different @@ -165,6 +176,26 @@ """ +# Appended in subagent mode only (when job_chat is called from global_chat). +subagent_mode_instructions = """ + +The workflow has other steps, but your edit_job tool edits THIS step only. +Decide by where the change lands: + +- Change lands in THIS step (or nothing needs changing): handle it here. Read + any other step with `inspect_job_code` whenever it helps — what an upstream + step outputs, keeping style or field names consistent, finding code the user + mentions that is not in . +- Change lands anywhere else — workflow structure (add/remove/rename steps, + triggers, edges, adaptors) or another step's code, even code the user calls + "this step": call `edit_workflow` as your very first action, before any + reply text (at most exactly: "I'll take a look at your workflow."). + +If the user mentions code you can't find in , assume it lives in +another step — never reply that it isn't here. + +""" + # Response contract, appended last in the system message. output_format = """ @@ -261,10 +292,37 @@ def has(self, key): return hasattr(self, key) and getattr(self, key) is not None -def generate_system_message(context_dict, search_results, download_adaptor_docs=True, stream_manager=None): +def build_focus_line(viewing, focused): + """One sentence orienting the model to what the user has on screen, and which + step it can edit when that differs. + + `viewing` (router-only) is the on-screen step name, the literal "canvas", or + None when the caller can't tell; `focused` is the editable step. When both + are a step and coincide (the common case) they fuse into one clause; they + diverge only when the request targets a step other than the one open. + + Returns "" when there is nothing worth stating (no viewing info): the model + works from and the tools, and we avoid narrating editing plumbing + the model could echo back to the user. + """ + if viewing and viewing != "canvas": + if not focused: + return f"The user has the '{viewing}' step's code open." + if normalize_name(viewing) == normalize_name(focused): + return f"The user has the '{viewing}' step's code open — that's the step you're currently editing." + return f"The user has the '{viewing}' step open, but the step you're editing is '{focused}' — likely what their request is about." + if viewing == "canvas": + if focused: + return f"The user is viewing the workflow canvas, not a specific step; the '{focused}' step is the one you're currently editing." + return "The user is viewing the workflow canvas." + return "" + + +def generate_system_message(context_dict, search_results, download_adaptor_docs=True, stream_manager=None, + workflow_yaml=None, subagent=False): context = context_dict if isinstance(context_dict, Context) else Context(**(context_dict or {})) - message = [system_role] + message = [subagent_system_role if subagent else system_role] message.append(f"{job_writing_summary}") message.append({"type": "text", "text": ".", "cache_control": {"type": "ephemeral"}}) @@ -353,6 +411,27 @@ def generate_system_message(context_dict, search_results, download_adaptor_docs= ```{context.log}``` """) + if subagent: + message.append(subagent_mode_instructions) + if workflow_yaml: + # `focused` is the editable step; `viewing` (router-only) is what the + # user actually has on screen — a step's code or the literal "canvas". + # Grounding focus in their view lets a bare "this step" resolve to + # what they're looking at. Both may be absent (planner/prod, or an + # unresolved page), in which case build_focus_line returns "". + focused = context.job_key if context.has("job_key") else ( + context.page_name if context.has("page_name") else None) + viewing = context.viewing if context.has("viewing") else None + focus = build_focus_line(viewing, focused) + header = ["The full workflow, job code redacted."] + if focus: + header.append(focus) + header.append("READ other steps' code with `inspect_job_code` when the request refers to them.") + redacted = redact_job_bodies(workflow_yaml) + message.append( + f"\n{' '.join(header)}\n\n{redacted}\n" + ) + # Output contract goes LAST so it is the final, most prominent instruction. message.append(output_format) @@ -365,7 +444,8 @@ def format_search_results(search_results): ]) @observe(name="job_chat_build_prompt") -def build_prompt(content, history, context, rag=None, api_key=None, stream_manager=None, download_adaptor_docs=True, refresh_rag=False): +def build_prompt(content, history, context, rag=None, api_key=None, stream_manager=None, download_adaptor_docs=True, refresh_rag=False, + workflow_yaml=None, subagent=False): retrieved_knowledge = { "search_results": [], "search_results_sections": [], @@ -398,7 +478,9 @@ def build_prompt(content, history, context, rag=None, api_key=None, stream_manag context_dict=context, search_results=retrieved_knowledge.get("search_results") if retrieved_knowledge is not None else None, download_adaptor_docs=download_adaptor_docs, - stream_manager=stream_manager) + stream_manager=stream_manager, + workflow_yaml=workflow_yaml, + subagent=subagent) prompt = [] prompt.extend(history) @@ -407,9 +489,15 @@ def build_prompt(content, history, context, rag=None, api_key=None, stream_manag # only remind the model to route an actual code change through `edit_job`. # Added only to the message sent to the model; the stored history (built in # generate() from the raw content) omits it, so it never accumulates. + reminder = "Reply in text. If this requires changing the job code, also call the `edit_job` tool to apply the change." + if subagent: + reminder += ( + " If it needs changes beyond this step's code, call `edit_workflow` first;" + " to merely read another step (to answer, or to edit this one), use `inspect_job_code`." + ) prompt.append({ "role": "user", - "content": f"{content}\n\nReply in text. If this requires changing the job code, also call the `edit_job` tool to apply the change.", + "content": f"{content}\n\n{reminder}", }) return (system_message, prompt, retrieved_knowledge) diff --git a/services/job_chat/tests/unit/test_subagent_prompt.py b/services/job_chat/tests/unit/test_subagent_prompt.py new file mode 100644 index 00000000..a75fc808 --- /dev/null +++ b/services/job_chat/tests/unit/test_subagent_prompt.py @@ -0,0 +1,77 @@ +"""Unit tests for job_chat's subagent-mode system prompt.""" + +from job_chat.prompt import generate_system_message + +WORKFLOW_YAML = """\ +name: wf +jobs: + fetch-patients: + name: Fetch Patients + body: get('/patients'); +""" + + +def system_text(**kwargs) -> str: + blocks = generate_system_message(context_dict={}, search_results=None, **kwargs) + return "\n".join(b["text"] for b in blocks) + + +def test_production_prompt_keeps_navigate_instruction(): + text = system_text() + + # Production callers never set subagent: the only-job-code scope and the + # go-to-the-workflow-overview instruction must stay untouched, and no + # subagent sections appear + assert "tell them to navigate to the workflow overview" in text + assert "Do NOT help with overall workflow structure" in text + assert "edit_workflow" not in text + assert "" not in text + + +def test_subagent_prompt_strips_navigate_instruction(): + text = system_text(subagent=True, workflow_yaml=WORKFLOW_YAML) + + # Neither the go-elsewhere phrasing nor the only-job-code scope may be in + # context at all (production_scope_instructions is omitted from the + # composed subagent_system_role) + assert "navigate to the workflow overview" not in text + assert "Do NOT help with overall workflow structure" not in text + assert "ONLY help with job code" not in text + assert "edit_workflow" in text + assert "" in text + + +def _subagent_text(context: dict) -> str: + blocks = generate_system_message( + context_dict=context, search_results=None, + subagent=True, workflow_yaml=WORKFLOW_YAML, + ) + return "\n".join(b["text"] for b in blocks) + + +def test_subagent_prompt_grounds_focus_in_viewed_step(): + # No viewing info: no focus sentence is emitted (the model works from + # ), and no editing-plumbing phrasing leaks. Structure still shown. + text = system_text(subagent=True, workflow_yaml=WORKFLOW_YAML) + assert "" in text + assert "No step is focused" not in text + assert "loaded for editing" not in text + assert "currently editing" not in text + + # Viewing the same step it can edit (common case): named and framed as open. + text = _subagent_text({"job_key": "fetch-patients", "viewing": "Fetch Patients"}) + assert "Fetch Patients" in text + assert "currently editing" in text + + # Viewing the workflow canvas: says "canvas" and still names the editable + # step, with no editing-plumbing phrase. + text = _subagent_text({"job_key": "fetch-patients", "viewing": "canvas"}) + assert "workflow canvas" in text + assert "'fetch-patients'" in text + assert "loaded for editing" not in text + + # Mismatch — viewing one step, editing another: states both, softly. + text = _subagent_text({"job_key": "fetch-patients", "viewing": "Notify Admin"}) + assert "Notify Admin" in text + assert "'fetch-patients'" in text + assert "likely what their request is about" in text diff --git a/services/workflow_chat/gen_project_prompt.py b/services/workflow_chat/gen_project_prompt.py index 36283edb..35eeaf29 100644 --- a/services/workflow_chat/gen_project_prompt.py +++ b/services/workflow_chat/gen_project_prompt.py @@ -30,17 +30,18 @@ def build_system_message(mode_config, existing_yaml=None): return system_message -def build_prompt(content, existing_yaml=None, errors=None, history=None, read_only=False): +def build_prompt(content, existing_yaml=None, errors=None, history=None, read_only=False, subagent=False): """ Build a prompt for the LLM based on mode and context. - + Args: content: User message content existing_yaml: Current YAML being edited (optional) errors: Error messages if in error mode (optional) history: Conversation history (optional) read_only: Whether in read-only mode - + subagent: Whether called from global_chat (adds handover instructions) + Returns: Tuple of (system_message, prompt_messages) """ @@ -75,8 +76,22 @@ def build_prompt(content, existing_yaml=None, errors=None, history=None, read_on user_content = content system_message = build_system_message(mode_config, existing_yaml) - + + if subagent: + # Job-code requests are handed over instead — remove the decline-and- + # navigate-to-the-Inspector instruction (it appears in two prompt + # sections) so it can never slip out. Must match prompts yaml verbatim; + # a unit test guards against the two drifting apart. + system_message = system_message.replace( + "If the user asks for job code, DECLINE to provide it yet, and explain that they " + "need to save their workflow and then navigate to the specific job's code page in " + "the Inspector. Once there, you can help them write the code (and will be able to " + "see any existing code for that job).", + 'If the user asks for job code, set "handover" (see Job Code Requests below).', + ) + system_message += "\n" + config_loader.get_prompt("subagent_handover_instructions") + prompt = list(history) # Create a copy prompt.append({"role": "user", "content": user_content}) - + return (system_message, prompt) \ No newline at end of file diff --git a/services/workflow_chat/gen_project_prompts.yaml b/services/workflow_chat/gen_project_prompts.yaml index 07da00e7..ae5b4ca7 100644 --- a/services/workflow_chat/gen_project_prompts.yaml +++ b/services/workflow_chat/gen_project_prompts.yaml @@ -276,3 +276,15 @@ prompts: Always set the "yaml" key to null. The user's latest message and prior conversation are provided below. Generate your response accordingly. + + subagent_handover_instructions: | + + ## Job Code Requests + + Your response JSON has an extra FIRST field: "handover". + - If the request is chiefly about the code inside a step (reading, explaining, + debugging, or editing it), set "handover" to a short reason, "yaml" to null and + "text" to "" — the request is then rerouted and handled with full code access. + Never decline such requests or tell the user to navigate elsewhere or save + first; this overrides earlier instructions. + - Otherwise set "handover" to null and answer normally. diff --git a/services/workflow_chat/tests/unit/client/test_handover.py b/services/workflow_chat/tests/unit/client/test_handover.py new file mode 100644 index 00000000..732c4d0c --- /dev/null +++ b/services/workflow_chat/tests/unit/client/test_handover.py @@ -0,0 +1,45 @@ +"""Unit tests for subagent-mode handover parsing in workflow_chat.""" + +import json + +from workflow_chat.workflow_chat import AnthropicClient + + +def make_client() -> AnthropicClient: + """Build an AnthropicClient without an API key.""" + client = AnthropicClient.__new__(AnthropicClient) + client._streamed_yaml = None + client._handover = None + return client + + +def test_split_captures_handover_and_skips_yaml() -> None: + client = make_client() + response = json.dumps({"handover": "asks about job code", "yaml": "name: wf", "text": ""}) + + text, output_yaml = client.split_format_yaml(response) + + assert client._handover == "asks about job code" + assert output_yaml == "" + assert text == "" + + +def test_split_without_handover_behaves_normally() -> None: + client = make_client() + response = json.dumps({"handover": None, "yaml": None, "text": "The trigger runs daily."}) + + text, output_yaml = client.split_format_yaml(response) + + assert client._handover is None + assert text == "The trigger runs daily." + assert output_yaml == "" + + +def test_split_legacy_schema_without_handover_field() -> None: + client = make_client() + response = json.dumps({"yaml": None, "text": "Answer."}) + + text, _ = client.split_format_yaml(response) + + assert client._handover is None + assert text == "Answer." diff --git a/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py b/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py index a848a180..b1500d7f 100644 --- a/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py +++ b/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py @@ -33,6 +33,36 @@ def test_build_prompt_error_mode(): assert prompt[-1]["content"] == "Fix the workflow\nThis is the error message:\nInvalid trigger type" +def test_build_prompt_production_keeps_inspector_instruction(): + system_msg, _ = build_prompt( + content="Create a workflow", + existing_yaml="name: test-workflow", + history=[], + ) + + # Production callers never set subagent: the decline-and-navigate + # instruction must stay untouched + assert "navigate to the specific job's code page in the Inspector" in system_msg + assert "handover" not in system_msg + + +def test_build_prompt_subagent_strips_inspector_instruction(): + system_msg, _ = build_prompt( + content="Create a workflow", + existing_yaml="name: test-workflow", + history=[], + subagent=True, + ) + + # The go-elsewhere phrasing must not be in context at all, in any of the + # prompt sections it appears in — if this fails, the sentence in + # gen_project_prompts.yaml and the replace() in build_prompt have drifted + assert "navigate to the specific job's code page in the Inspector" not in system_msg + assert "DECLINE" not in system_msg + assert 'If the user asks for job code, set "handover"' in system_msg + assert "Job Code Requests" in system_msg + + def test_build_prompt_readonly_mode(): system_msg, prompt = build_prompt( content="What does this workflow do?", diff --git a/services/workflow_chat/workflow_chat.py b/services/workflow_chat/workflow_chat.py index a4cf9366..67ee6711 100644 --- a/services/workflow_chat/workflow_chat.py +++ b/services/workflow_chat/workflow_chat.py @@ -29,6 +29,25 @@ "required": ["yaml", "text"], "additionalProperties": False } + +# Subagent mode (called from global_chat): adds a "handover" field so the model +# can hand a misrouted request back to the caller. It comes FIRST so it is +# generated before yaml/text — streaming can then suppress output and the +# router reroutes before the user sees anything. +_SUBAGENT_OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "handover": { + "anyOf": [ + {"type": "string"}, + {"type": "null"} + ] + }, + **_OUTPUT_SCHEMA["properties"] + }, + "required": ["handover", "yaml", "text"], + "additionalProperties": False +} from anthropic import ( Anthropic, APIConnectionError, @@ -89,6 +108,9 @@ class Payload: stream: Optional[bool] = False read_only: Optional[bool] = False metrics_opt_in: Optional[bool] = None + # Subagent mode: set only when called from global_chat, never by direct + # production callers. Enables the handover response field. + subagent: Optional[bool] = False @classmethod def from_dict(cls, data: Dict[str, Any]) -> "Payload": @@ -107,6 +129,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "Payload": stream=data.get("stream", False), read_only=data.get("read_only", False), metrics_opt_in=data.get("metrics_opt_in"), + subagent=data.get("subagent", False), ) @@ -123,6 +146,8 @@ class ChatResponse: content_yaml: str history: List[Dict[str, str]] usage: Dict[str, Any] + # Subagent mode only: reason the request was handed back to the caller + handover: Optional[str] = None class AnthropicClient: @@ -137,6 +162,9 @@ def __init__(self, config: Optional[ChatConfig] = None): # so restore_components runs once and new-component UUIDs stay identical # between the streamed preview and the persisted payload. self._streamed_yaml = None + # Subagent mode: handover reason parsed from the model's response. + # Set as early as possible while streaming so text output is suppressed. + self._handover = None @staticmethod def _unescape_json_string(text): @@ -160,13 +188,15 @@ def generate( stream: Optional[bool] = False, current_page: Optional[dict] = None, read_only: Optional[bool] = False, + subagent: Optional[bool] = False, + stream_manager: Optional[StreamManager] = None, ) -> ChatResponse: """Generate a response using the Claude API. Retry up to 2 times if YAML/JSON parsing fails.""" - + with sentry_sdk.start_transaction(name="workflow_generation") as transaction: history = history.copy() if history else [] - stream_manager = StreamManager(model=self.config.model, stream=stream) + stream_manager = stream_manager or StreamManager(model=self.config.model, stream=stream) # Extract and preserve existing components (skip in read-only mode) preserved_values = {} @@ -189,14 +219,15 @@ def generate( existing_yaml=processed_existing_yaml, errors=errors, history=history, - read_only=read_only + read_only=read_only, + subagent=subagent ) # Structured outputs config — guarantees valid JSON matching schema output_config = { "format": { "type": "json_schema", - "schema": _OUTPUT_SCHEMA + "schema": _SUBAGENT_OUTPUT_SCHEMA if subagent else _OUTPUT_SCHEMA }, "effort": "medium" } @@ -212,6 +243,7 @@ def generate( for attempt in range(max_retries + 1): # Reset per attempt so a retry never reuses a prior stream's YAML self._streamed_yaml = None + self._handover = None with sentry_sdk.start_span(description="anthropic_api_call"): if stream: logger.info("Making streaming API call") @@ -244,7 +276,7 @@ def generate( message = stream_obj.get_final_message() # Flush any remaining buffered text, stripping JSON closing chars - if text_started: + if text_started and not self._handover: if sent_length < len(accumulated_response): remaining = accumulated_response[sent_length:] remaining = re.sub(r'"\s*}\s*$', '', remaining) @@ -279,6 +311,18 @@ def generate( # If YAML parsing succeeded or we're on the last attempt, return the result if response_yaml is not None or attempt == max_retries: + if self._handover: + logger.info(f"workflow_chat handing over: {self._handover}") + # Deliberately do NOT end the stream: the caller reroutes + # the request and the next agent continues on the same stream. + return ChatResponse( + content=response_text or "", + content_yaml=None, + history=history, + usage=accumulated_usage, + handover=self._handover, + ) + if not response_text: stop_reason = getattr(message, "stop_reason", None) empty_reason = "max_tokens" if stop_reason == "max_tokens" else "no_text_blocks" @@ -427,6 +471,12 @@ def split_format_yaml(self, response, preserved_values=None, stream_manager=None # Try to parse the response as JSON response_data = json.loads(response) + # Subagent mode: a handover means the request is being handed back + # to the caller — capture the reason and skip the YAML entirely + if response_data.get("handover"): + self._handover = response_data["handover"] + return response_data.get("text", "").strip(), "" + # Extract text and yaml from the JSON output_text = response_data.get("text", "").strip() raw_yaml = response_data.get("yaml") or "" @@ -600,15 +650,25 @@ def process_stream_event(self, event, accumulated_response, text_started, sent_l match = re.search(r'"text"\s*:\s*"', accumulated_response) if match: - # Close the partial object and extract the yaml field + # Close the partial object and extract the fields + # generated before "text" (yaml, and in subagent mode + # the handover reason, which comes first) yaml_part = accumulated_response[:match.start()] yaml_raw = yaml_part.rstrip().rstrip(",") + "}" try: - yaml_value = json.loads(yaml_raw).get("yaml") - except (json.JSONDecodeError, ValueError, AttributeError): - yaml_value = None - - if yaml_value: + partial = json.loads(yaml_raw) + except (json.JSONDecodeError, ValueError): + partial = None + if not isinstance(partial, dict): + partial = {} + + if partial.get("handover"): + # Handed back to the caller: suppress all output — + # the rerouted agent produces the user-facing reply + self._handover = partial["handover"] + + yaml_value = partial.get("yaml") + if yaml_value and not self._handover: # Finalize before sending so the streamed preview carries # real IDs/code, not raw placeholders. Cache it so the final # response reuses the identical YAML. Only send if the content @@ -626,7 +686,7 @@ def process_stream_event(self, event, accumulated_response, text_started, sent_l sent_length = match.end() text_started = True - if text_started: + if text_started and not self._handover: # Text phase: stream with buffer for split escape sequences buffer_size = 2 safe_to_send_until = len(accumulated_response) - buffer_size @@ -646,7 +706,7 @@ def main(data_dict: dict) -> dict: """ try: sentry_sdk.set_context("request_data", { - k: v for k, v in data_dict.items() if k != "api_key" + k: v for k, v in data_dict.items() if k not in ("api_key", "_stream_manager") }) data = Payload.from_dict(data_dict) @@ -686,7 +746,11 @@ def main(data_dict: dict) -> dict: history=data.history, stream=data.stream, current_page=current_page, - read_only=data.read_only + read_only=data.read_only, + subagent=data.subagent, + # In-process callers (global_chat) may inject a shared stream + # manager so a handed-over request continues the same stream + stream_manager=data_dict.get("_stream_manager"), ) if tracking: @@ -698,6 +762,12 @@ def main(data_dict: dict) -> dict: if diff_meta: langfuse.update_current_span(metadata=diff_meta) + # Tag the trace when the request was handed back for rerouting to + # the planner, so we can filter for handovers. + if tracking and result.handover: + with propagate_attributes(tags=["handover"]): + pass + # Build response response_dict = { "response": result.content, @@ -707,6 +777,9 @@ def main(data_dict: dict) -> dict: "meta": {"apollo_version": APOLLO_VERSION} } + if result.handover: + response_dict["handover"] = result.handover + return response_dict except ApolloError: diff --git a/services/yaml_utils.py b/services/yaml_utils.py new file mode 100644 index 00000000..17920ec7 --- /dev/null +++ b/services/yaml_utils.py @@ -0,0 +1,202 @@ +""" +Shared utility functions for working with workflow YAML strings. + +Used by global_chat (router, planner, subagent caller) and by job_chat in +subagent mode for job extraction, code stitching, and step inspection. +""" +import re + +import yaml + + +def get_page_view(page: str | None) -> tuple[str | None, str | None]: + """ + Classify what the user has on screen from the `page` breadcrumb — the single + parser for that URL (get_step_name_from_page delegates to this). + + Shapes (names are raw, may contain spaces): + workflows// -> ("step", "") job code page + workflows/ -> ("overview", None) workflow canvas + settings / absent / anything else -> (None, None) + + Because a name may itself contain "/", the returned step name is a + best-effort candidate — the caller must validate it against the workflow + YAML rather than trust it. + """ + if not page: + return None, None + parts = page.strip("/").split("/") + if parts[0] != "workflows": + return None, None + if len(parts) == 2: + return "overview", None + if len(parts) == 3 and parts[2] != "settings": + return "step", parts[2] + return None, None + + +def get_step_name_from_page(page: str | None) -> str | None: + """ + Extract the focused step name from a job-code page URL, or None for the + canvas, settings, or an unrecognized value. + + Examples: + workflows/my-workflow/fetch-patients -> "fetch-patients" + workflows/my-workflow -> None + workflows/my-workflow/settings -> None + """ + view, step = get_page_view(page) + return step if view == "step" else None + + +def normalize_name(name: str) -> str: + """Normalize a name for fuzzy matching: lowercase, non-alphanumeric chars become hyphens.""" + return re.sub(r'[^a-z0-9]', '-', name.lower()).strip('-') + + +def find_job_in_yaml(yaml_str: str, step_name: str) -> tuple[str | None, dict | None]: + """ + Find a job in the workflow YAML by step name. + + Tries direct key match first, then normalized name comparison against + both the job key and the job's name field. + + Returns: + (job_key, job_data) or (None, None) if not found or on parse error + """ + try: + yaml_data = yaml.safe_load(yaml_str) + except Exception: + return None, None + + if not yaml_data or "jobs" not in yaml_data: + return None, None + + jobs = yaml_data["jobs"] + + # Direct key match + if step_name in jobs: + return step_name, jobs[step_name] + + # Normalized match: compare against job key and name field + normalized_step = normalize_name(step_name) + for job_key, job_data in jobs.items(): + if normalize_name(job_key) == normalized_step: + return job_key, job_data + job_name = job_data.get("name", "") + if normalize_name(job_name) == normalized_step: + return job_key, job_data + + return None, None + + +EMPTY_JOB_BODY = "// Add operations here" + + +def workflow_has_job_code(yaml_str: str | None) -> bool: + """Return True if any job has a non-empty, non-placeholder body. + + The canonical empty-job marker is ``// Add operations here`` (see + workflow_chat); a blank body or that marker means "no code yet". Used to + decide whether a "what does this do" question needs the planner (to read the + real code) or can take the faster workflow_agent path (structure only). + """ + try: + yaml_data = yaml.safe_load(yaml_str) + except Exception: + return False + if not yaml_data or "jobs" not in yaml_data: + return False + for job_data in yaml_data["jobs"].values(): + body = (job_data or {}).get("body") + if isinstance(body, str) and body.strip() and body.strip() != EMPTY_JOB_BODY: + return True + return False + + +def redact_job_bodies(yaml_str: str) -> str: + """Return workflow YAML with job bodies replaced by a placeholder and id + fields removed. + + This is the read-only structural view shown to the planner and to job_chat + in subagent mode. It never round-trips back into a real workflow, so the + UUID ids are pure noise to the model — dropping them saves tokens. + """ + try: + yaml_data = yaml.safe_load(yaml_str) + if yaml_data and "jobs" in yaml_data: + _remove_ids(yaml_data) + for job_data in yaml_data["jobs"].values(): + if "body" in job_data: + job_data["body"] = "# [use inspect_job_code to view]" + return yaml.dump(yaml_data, sort_keys=False) + except Exception: + pass + return yaml_str + + +def _remove_ids(obj: object) -> None: + """Recursively remove 'id' keys from a parsed YAML structure.""" + if isinstance(obj, dict): + obj.pop("id", None) + for value in obj.values(): + _remove_ids(value) + elif isinstance(obj, list): + for item in obj: + _remove_ids(item) + + +def stitch_job_code(yaml_str: str, job_key: str, new_code: str) -> str: + """ + Replace a job's body in the workflow YAML with new code. + + Returns the original YAML string unchanged if parsing or stitching fails. + """ + try: + yaml_data = yaml.safe_load(yaml_str) + if yaml_data and "jobs" in yaml_data and job_key in yaml_data["jobs"]: + yaml_data["jobs"][job_key]["body"] = new_code + return yaml.dump(yaml_data, sort_keys=False) + except Exception: + pass + + return yaml_str + + +# Read-only step inspection, shared by the planner and job_chat (subagent +# mode) so both agents explore the workflow with the exact same tool. + +INSPECT_JOB_CODE_TOOL = { + "name": "inspect_job_code", + "description": """Read the current code body of one or more jobs in the workflow (read-only). + +Use this to inspect existing step code before editing — e.g. to find which steps a change applies to before editing only those, or to base one step on another. Pass all the job keys you need in a single call rather than calling once per job.""", + "input_schema": { + "type": "object", + "properties": { + "job_keys": { + "type": "array", + "items": {"type": "string"}, + "description": "The job keys to inspect (e.g. ['fetch-patients', 'load-dhis2'])", + }, + }, + "required": ["job_keys"], + }, +} + + +def inspect_job_code(yaml_str: str | None, job_keys: list[str]) -> str: + """Execute the inspect_job_code tool: return the named jobs' code bodies.""" + if not yaml_str: + return "No workflow available to inspect." + if not job_keys: + return "ERROR: No job keys provided." + + parts = [] + for job_key in job_keys: + _, job_data = find_job_in_yaml(yaml_str, job_key) + if job_data and job_data.get("body"): + parts.append(f"Job code for '{job_key}':\n\n{job_data['body']}") + else: + parts.append(f"No code found for job '{job_key}'.") + return "\n\n".join(parts) From bb87c23a0d0a4fb6f23b525fff2ef441115a6a14 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Thu, 30 Jul 2026 17:58:40 +0100 Subject: [PATCH 5/5] version: 3.1.0 --- .changeset/five-bags-rule.md | 6 ------ .changeset/short-months-look.md | 6 ------ CHANGELOG.md | 12 ++++++++++++ package.json | 2 +- 4 files changed, 13 insertions(+), 13 deletions(-) delete mode 100644 .changeset/five-bags-rule.md delete mode 100644 .changeset/short-months-look.md diff --git a/.changeset/five-bags-rule.md b/.changeset/five-bags-rule.md deleted file mode 100644 index d8c883b6..00000000 --- a/.changeset/five-bags-rule.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -"apollo": patch ---- - -global_chat: enable subagents to pull missing context, recovering from routing -errors diff --git a/.changeset/short-months-look.md b/.changeset/short-months-look.md deleted file mode 100644 index 2dcff69f..00000000 --- a/.changeset/short-months-look.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -"apollo": minor ---- - -global chat: add answer streaming to the planner, breaking up responses into -chunks which can be rendered earlier diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f74eb1b..3154ccfd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # apollo +## 3.1.0 + +### Minor Changes + +- f1df29c: global chat: add answer streaming to the planner, breaking up + responses into chunks which can be rendered earlier + +### Patch Changes + +- 6c322ae: global_chat: enable subagents to pull missing context, recovering + from routing errors + ## 3.0.3 ### Patch Changes diff --git a/package.json b/package.json index 92c7df1a..8f2790bb 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "apollo", "module": "platform/index.ts", - "version": "3.0.3", + "version": "3.1.0", "type": "module", "scripts": { "start": "NODE_ENV=production bun platform/src/index.ts",