From 79d92459059dff6e5770f1012898220608f37017 Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Thu, 16 Jul 2026 09:28:29 +0100 Subject: [PATCH 01/11] add subagent mode and inspect tool --- services/global_chat/planner.py | 28 +- services/global_chat/router.py | 59 +++- services/global_chat/subagent_caller.py | 2 +- .../global_chat/tests/unit/test_router.py | 65 ++++- .../global_chat/tests/unit/test_yaml_utils.py | 35 +++ .../global_chat/tools/tool_definitions.py | 21 +- services/job_chat/job_chat.py | 272 +++++++++++++----- services/job_chat/prompt.py | 65 ++++- services/workflow_chat/gen_project_prompt.py | 14 +- .../workflow_chat/gen_project_prompts.yaml | 17 ++ .../tests/unit/client/test_handover.py | 45 +++ services/workflow_chat/workflow_chat.py | 95 +++++- services/{global_chat => }/yaml_utils.py | 52 +++- 13 files changed, 628 insertions(+), 142 deletions(-) create mode 100644 services/global_chat/tests/unit/test_yaml_utils.py create mode 100644 services/workflow_chat/tests/unit/client/test_handover.py rename services/{global_chat => }/yaml_utils.py (64%) diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index 5fec69ff..45a7b472 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,23 @@ 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") +<<<<<<< HEAD +======= + 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) + +>>>>>>> d8fec45 (add subagent mode and inspect tool) self.current_yaml = workflow_yaml self.yaml_modified = False self._user = user @@ -488,19 +500,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/router.py b/services/global_chat/router.py index db0bdcf3..2294f25c 100644 --- a/services/global_chat/router.py +++ b/services/global_chat/router.py @@ -19,9 +19,10 @@ from langfuse import observe 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, 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) @@ -122,7 +127,7 @@ def route_and_execute( decision = RouterDecision(destination="planner", confidence=1) 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 +234,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 +252,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 +343,9 @@ 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 clean_history = [{"role": t["role"], "content": t["content"]} for t in history] enriched_content = self._format_attachments_for_content(content) @@ -343,9 +359,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 +394,33 @@ 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") + + 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", {})) + planner_result.meta["handover_from"] = from_agent + planner_result.meta["handover_reason"] = reason + return planner_result + def _route_to_planner( self, content: str, @@ -395,6 +447,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/unit/test_router.py b/services/global_chat/tests/unit/test_router.py index 1a3ada34..f4911b73 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, RouterResult +from yaml_utils import workflow_has_job_code EMPTY_YAML = """\ name: wf @@ -33,6 +33,7 @@ def make_router() -> RouterAgent: router._input_attachments = [] router._user = None router._metrics_opt_in = None + router._stream_manager = None return router @@ -87,3 +88,63 @@ 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", + 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" + assert result.meta["handover_from"] == "job_code_agent" + assert result.meta["handover_reason"] == "needs structure changes" + # 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 result.meta["handover_from"] == "workflow_agent" + assert result.usage["input_tokens"] == 13 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..db16e5ec --- /dev/null +++ b/services/global_chat/tests/unit/test_yaml_utils.py @@ -0,0 +1,35 @@ +"""Unit tests for the shared inspect_job_code tool executor.""" + +from yaml_utils import inspect_job_code + +WORKFLOW_YAML = """\ +name: wf +jobs: + fetch-patients: + name: Fetch Patients + body: get('/patients'); + send-data: + name: Send Data + body: post('/data', $.data); +""" + + +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/job_chat/job_chat.py b/services/job_chat/job_chat.py index 8a52f4fd..28802549 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,35 @@ }, } +# Subagent mode only (job_chat called from global_chat): hand the request back +# to the caller, which reroutes it to an agent that can complete it. +_HANDOVER_TOOL = { + "name": "handover", + "description": ( + "Hand this request over to a more capable assistant. Call this — as your " + "VERY FIRST action, with no reply text before it — when the request is not " + "something you can complete from here: it is chiefly about workflow " + "structure (adding/removing/reordering steps, triggers, edges, adaptors), " + "or it needs code changes in a step other than the focused one or in " + "several steps. Never apologise for missing context; hand over instead." + ), + "input_schema": { + "type": "object", + "properties": { + "reason": { + "type": "string", + "description": "One sentence explaining what the request needs", + } + }, + "required": ["reason"], + "additionalProperties": False, + }, +} + +# 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 +146,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 +170,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 +190,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 +225,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 +259,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 +279,141 @@ 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(_HANDOVER_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 + 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) + + if hasattr(message, "usage"): + usage_events.append(message.usage.model_dump()) + + 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"] + + handover_block = next((b for b in tool_uses if b.name == "handover"), None) + if handover_block: + handover_reason = (handover_block.input or {}).get("reason") 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": + result_text = inspect_job_code(workflow_yaml, (block.input or {}).get("job_keys") or []) + else: + result_text = ( + "Not applied. Finish inspecting, then write your final reply " + "and call edit_job again with the complete edits." ) - 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=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) + 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 +422,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 +469,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 +722,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 +781,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. @@ -694,6 +813,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..35c19d40 100644 --- a/services/job_chat/prompt.py +++ b/services/job_chat/prompt.py @@ -4,6 +4,7 @@ import sentry_sdk from langfuse import observe from util import create_logger, ApolloError, AdaptorSpecifier, get_db_connection +from yaml_utils import redact_job_bodies from .retrieve_docs import retrieve_knowledge from search_adaptor_docs.search_adaptor_docs import fetch_signatures @@ -165,6 +166,28 @@ """ +# Appended in subagent mode only (when job_chat is called from global_chat). +subagent_mode_instructions = """ + +You are the job-code specialist inside a single unified OpenFn assistant. The +user sees one assistant, so never mention routing, agents, or internal +mechanics. NEVER say you cannot see the workflow, a step, or the job code, and +never tell the user to navigate to another page or paste something in — this +overrides any earlier instruction to send the user to the workflow overview. + +If the request is not something you can complete from here, call the +`handover` tool as your VERY FIRST action, before writing any reply text, with +a one-sentence reason. Hand over when the request: +- is chiefly about workflow structure (adding, removing, renaming, or + reordering steps; triggers; edges; changing adaptors), or +- needs code changes in a step other than the focused one, or in several steps. + +Otherwise answer as normal. You may READ other steps' code with the +`inspect_job_code` tool (when available) to answer questions that reference +them — do this instead of saying you can't see a step. + +""" + # Response contract, appended last in the system message. output_format = """ @@ -261,7 +284,8 @@ 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 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] @@ -353,6 +377,30 @@ def generate_system_message(context_dict, search_results, download_adaptor_docs= ```{context.log}``` """) + if subagent: + message.append(subagent_mode_instructions) + if workflow_yaml: + focused = context.job_key if context.has("job_key") else ( + context.page_name if context.has("page_name") else None) + if focused: + focused_line = ( + f"The focused step — the one belongs to and the ONLY one " + f"you can edit — is '{focused}'." + ) + else: + focused_line = ( + "No focused step could be resolved for this request. You may READ any " + "step with `inspect_job_code` to answer questions, but code edits " + "cannot be applied — if the user wants code changed, call `handover`." + ) + redacted = redact_job_bodies(workflow_yaml) + message.append( + f"\nThe user's full workflow is shown below with job " + f"code bodies redacted. {focused_line} Use the `inspect_job_code` tool to " + f"read the code of other steps when the request refers to them.\n\n" + f"{redacted}\n" + ) + # Output contract goes LAST so it is the final, most prominent instruction. message.append(output_format) @@ -365,7 +413,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 +447,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 +458,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 references other steps, read them with `inspect_job_code`; if it is" + " not about this step's code at all, call `handover` first instead of replying." + ) 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/workflow_chat/gen_project_prompt.py b/services/workflow_chat/gen_project_prompt.py index 36283edb..2de2b435 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,11 @@ 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: + 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..ee320427 100644 --- a/services/workflow_chat/gen_project_prompts.yaml +++ b/services/workflow_chat/gen_project_prompts.yaml @@ -276,3 +276,20 @@ 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: | + + ## Subagent Mode + + You are the workflow-structure specialist inside a single unified OpenFn assistant. + The user sees one assistant, so never mention routing, agents, or internal mechanics, + and never tell the user to navigate to another page or save and come back later — + this overrides any earlier instruction to do so. + + 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 job code / adaptor functions), do NOT answer it yourself + and do NOT decline: set "handover" to a one-sentence reason (naming the step if + you can), set "yaml" to null, and set "text" to "". The request will be handled + by an assistant that can see and edit job code. + - 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/workflow_chat.py b/services/workflow_chat/workflow_chat.py index a4cf9366..3eb8f890 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: @@ -707,6 +771,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/global_chat/yaml_utils.py b/services/yaml_utils.py similarity index 64% rename from services/global_chat/yaml_utils.py rename to services/yaml_utils.py index 119bbcf0..f6a6189b 100644 --- a/services/global_chat/yaml_utils.py +++ b/services/yaml_utils.py @@ -1,14 +1,15 @@ """ -Shared YAML utility functions for working with workflow YAML strings. +Shared utility functions for working with workflow YAML strings. -Used by router and subagent caller for job extraction and code stitching. +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 -from typing import Dict, Optional, Tuple -def get_step_name_from_page(page: Optional[str]) -> Optional[str]: +def get_step_name_from_page(page: str | None) -> str | None: """ Extract step name from page URL. @@ -32,7 +33,7 @@ def normalize_name(name: str) -> str: 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]]: +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. @@ -71,7 +72,7 @@ def find_job_in_yaml(yaml_str: str, step_name: str) -> Tuple[Optional[str], Opti EMPTY_JOB_BODY = "// Add operations here" -def workflow_has_job_code(yaml_str: Optional[str]) -> bool: +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 @@ -121,3 +122,42 @@ def stitch_job_code(yaml_str: str, job_key: str, new_code: str) -> str: 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 467bdffd2db2da8abe15a3b7ae66d7312c12ccd6 Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Thu, 16 Jul 2026 20:47:55 +0100 Subject: [PATCH 02/11] edits and tests --- services/global_chat/router.py | 29 +++- .../global_chat/tests/integration/__init__.py | 0 .../tests/integration/test_handover.py | 136 ++++++++++++++++++ .../global_chat/tests/unit/test_router.py | 38 ++++- .../global_chat/tests/unit/test_yaml_utils.py | 37 ++++- services/job_chat/job_chat.py | 49 ++++--- services/job_chat/prompt.py | 70 ++++----- .../tests/unit/test_subagent_prompt.py | 53 +++++++ services/workflow_chat/gen_project_prompt.py | 11 ++ .../workflow_chat/gen_project_prompts.yaml | 17 +-- .../unit/gen_project/test_prompt_build.py | 30 ++++ services/yaml_utils.py | 20 ++- 12 files changed, 420 insertions(+), 70 deletions(-) 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/job_chat/tests/unit/test_subagent_prompt.py diff --git a/services/global_chat/router.py b/services/global_chat/router.py index 2294f25c..771b936b 100644 --- a/services/global_chat/router.py +++ b/services/global_chat/router.py @@ -17,7 +17,7 @@ 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 @@ -126,6 +126,16 @@ 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, page, history, stream, decision.confidence) elif decision.destination == "job_code_agent": @@ -414,13 +424,26 @@ def _handover_to_planner( """ 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", {})) - planner_result.meta["handover_from"] = from_agent - planner_result.meta["handover_reason"] = reason 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, 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 f4911b73..ca514ec9 100644 --- a/services/global_chat/tests/unit/test_router.py +++ b/services/global_chat/tests/unit/test_router.py @@ -2,7 +2,7 @@ from unittest.mock import patch -from global_chat.router import RouterAgent, RouterResult +from global_chat.router import RouterAgent, RouterDecision, RouterResult from yaml_utils import workflow_has_job_code EMPTY_YAML = """\ @@ -34,6 +34,7 @@ def make_router() -> RouterAgent: router._user = None router._metrics_opt_in = None router._stream_manager = None + router.model = "claude-test" return router @@ -126,8 +127,8 @@ def test_job_route_handover_reroutes_to_planner() -> None: planner_mock.assert_called_once() assert result.response == "planner answer" - assert result.meta["handover_from"] == "job_code_agent" - assert result.meta["handover_reason"] == "needs structure changes" + # 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 @@ -146,5 +147,34 @@ def test_workflow_route_handover_reroutes_to_planner() -> None: ) planner_mock.assert_called_once() - assert result.meta["handover_from"] == "workflow_agent" + 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", 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 index db16e5ec..024965f4 100644 --- a/services/global_chat/tests/unit/test_yaml_utils.py +++ b/services/global_chat/tests/unit/test_yaml_utils.py @@ -1,6 +1,6 @@ -"""Unit tests for the shared inspect_job_code tool executor.""" +"""Unit tests for the shared inspect_job_code tool executor and redaction.""" -from yaml_utils import inspect_job_code +from yaml_utils import inspect_job_code, redact_job_bodies WORKFLOW_YAML = """\ name: wf @@ -13,6 +13,39 @@ 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"]) diff --git a/services/job_chat/job_chat.py b/services/job_chat/job_chat.py index 28802549..5bd929b9 100644 --- a/services/job_chat/job_chat.py +++ b/services/job_chat/job_chat.py @@ -83,31 +83,46 @@ }, } -# Subagent mode only (job_chat called from global_chat): hand the request back -# to the caller, which reroutes it to an agent that can complete it. -_HANDOVER_TOOL = { - "name": "handover", +# 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. +_INSPECT_WORKFLOW_TOOL = { + "name": "inspect_workflow", "description": ( - "Hand this request over to a more capable assistant. Call this — as your " - "VERY FIRST action, with no reply text before it — when the request is not " - "something you can complete from here: it is chiefly about workflow " - "structure (adding/removing/reordering steps, triggers, edges, adaptors), " - "or it needs code changes in a step other than the focused one or in " - "several steps. Never apologise for missing context; hand over instead." + "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, " + "with no reply text before it. To merely READ another step's code, use " + "inspect_job_code instead." ), "input_schema": { "type": "object", "properties": { - "reason": { + "goal": { "type": "string", - "description": "One sentence explaining what the request needs", + "description": "One sentence: what needs to be done", } }, - "required": ["reason"], + "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 inspect_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 @@ -283,9 +298,9 @@ def generate( if suggest_code: tools.append(_EDIT_TOOL) if subagent: - tools.append(_HANDOVER_TOOL) + tools.append(_INSPECT_WORKFLOW_TOOL) if workflow_yaml: - tools.append(INSPECT_JOB_CODE_TOOL) + 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 @@ -370,9 +385,9 @@ def generate( tool_uses = [b for b in message.content if getattr(b, "type", None) == "tool_use"] - handover_block = next((b for b in tool_uses if b.name == "handover"), None) + handover_block = next((b for b in tool_uses if b.name == "inspect_workflow"), None) if handover_block: - handover_reason = (handover_block.input or {}).get("reason") or "handover requested" + 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"] diff --git a/services/job_chat/prompt.py b/services/job_chat/prompt.py index 35c19d40..4092c154 100644 --- a/services/job_chat/prompt.py +++ b/services/job_chat/prompt.py @@ -10,7 +10,7 @@ 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. @@ -33,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 inspect_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. @@ -61,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 @@ -168,24 +178,25 @@ # Appended in subagent mode only (when job_chat is called from global_chat). subagent_mode_instructions = """ - -You are the job-code specialist inside a single unified OpenFn assistant. The -user sees one assistant, so never mention routing, agents, or internal -mechanics. NEVER say you cannot see the workflow, a step, or the job code, and -never tell the user to navigate to another page or paste something in — this -overrides any earlier instruction to send the user to the workflow overview. - -If the request is not something you can complete from here, call the -`handover` tool as your VERY FIRST action, before writing any reply text, with -a one-sentence reason. Hand over when the request: -- is chiefly about workflow structure (adding, removing, renaming, or - reordering steps; triggers; edges; changing adaptors), or -- needs code changes in a step other than the focused one, or in several steps. - -Otherwise answer as normal. You may READ other steps' code with the -`inspect_job_code` tool (when available) to answer questions that reference -them — do this instead of saying you can't see a step. - + +NEVER say you cannot see the workflow, a step, or its code; never mention your +tools or access; never send the user to another page. + +`edit_job` edits THIS step only. Changes to anything else — workflow structure +(add/remove/rename steps, triggers, edges, adaptors) or another step's code — +go through `inspect_workflow`: call it as your very first action, with no +reply needed (at most: "I'll take a look at your workflow."). + +If the user mentions code that is not in (a warning, function, or +behavior you can't find — even if they say "this step"), they might be describing +another step. Never reply that it isn't in this step: to +change it, call `inspect_workflow`; to answer a question about it, read it +with `inspect_job_code`. + +Use `inspect_job_code` to read other steps whenever it helps — e.g. to see +what an upstream step outputs, or to keep style, patterns, or field names +consistent across steps. + """ # Response contract, appended last in the system message. @@ -288,7 +299,7 @@ def generate_system_message(context_dict, search_results, download_adaptor_docs= 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"}}) @@ -383,21 +394,16 @@ def generate_system_message(context_dict, search_results, download_adaptor_docs= focused = context.job_key if context.has("job_key") else ( context.page_name if context.has("page_name") else None) if focused: - focused_line = ( - f"The focused step — the one belongs to and the ONLY one " - f"you can edit — is '{focused}'." - ) + focused_line = f"The focused step — the only one whose code you can edit — is '{focused}'." else: focused_line = ( - "No focused step could be resolved for this request. You may READ any " - "step with `inspect_job_code` to answer questions, but code edits " - "cannot be applied — if the user wants code changed, call `handover`." + "No step is focused, so code edits cannot be applied here — " + "for code changes call `inspect_workflow`." ) redacted = redact_job_bodies(workflow_yaml) message.append( - f"\nThe user's full workflow is shown below with job " - f"code bodies redacted. {focused_line} Use the `inspect_job_code` tool to " - f"read the code of other steps when the request refers to them.\n\n" + f"\nThe full workflow, job code redacted. {focused_line} " + f"READ other steps' code with `inspect_job_code` when the request refers to them.\n\n" f"{redacted}\n" ) @@ -461,8 +467,8 @@ def build_prompt(content, history, context, rag=None, api_key=None, stream_manag 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 references other steps, read them with `inspect_job_code`; if it is" - " not about this step's code at all, call `handover` first instead of replying." + " If it needs changes beyond this step's code, call `inspect_workflow` first;" + " to merely read another step (to answer, or to edit this one), use `inspect_job_code`." ) prompt.append({ "role": "user", 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..2dcf9d6a --- /dev/null +++ b/services/job_chat/tests/unit/test_subagent_prompt.py @@ -0,0 +1,53 @@ +"""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 "inspect_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 "inspect_workflow" in text + assert "" in text + + +def test_subagent_prompt_names_focused_step(): + text = system_text(subagent=True, workflow_yaml=WORKFLOW_YAML) + assert "No step is focused" in text + + blocks = generate_system_message( + context_dict={"job_key": "fetch-patients"}, search_results=None, + subagent=True, workflow_yaml=WORKFLOW_YAML, + ) + text = "\n".join(b["text"] for b in blocks) + assert "'fetch-patients'" in text diff --git a/services/workflow_chat/gen_project_prompt.py b/services/workflow_chat/gen_project_prompt.py index 2de2b435..35eeaf29 100644 --- a/services/workflow_chat/gen_project_prompt.py +++ b/services/workflow_chat/gen_project_prompt.py @@ -78,6 +78,17 @@ def build_prompt(content, existing_yaml=None, errors=None, history=None, read_on 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 diff --git a/services/workflow_chat/gen_project_prompts.yaml b/services/workflow_chat/gen_project_prompts.yaml index ee320427..ae5b4ca7 100644 --- a/services/workflow_chat/gen_project_prompts.yaml +++ b/services/workflow_chat/gen_project_prompts.yaml @@ -279,17 +279,12 @@ prompts: subagent_handover_instructions: | - ## Subagent Mode - - You are the workflow-structure specialist inside a single unified OpenFn assistant. - The user sees one assistant, so never mention routing, agents, or internal mechanics, - and never tell the user to navigate to another page or save and come back later — - this overrides any earlier instruction to do so. + ## 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 job code / adaptor functions), do NOT answer it yourself - and do NOT decline: set "handover" to a one-sentence reason (naming the step if - you can), set "yaml" to null, and set "text" to "". The request will be handled - by an assistant that can see and edit job code. + - 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/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/yaml_utils.py b/services/yaml_utils.py index f6a6189b..b993ecc3 100644 --- a/services/yaml_utils.py +++ b/services/yaml_utils.py @@ -94,10 +94,17 @@ def workflow_has_job_code(yaml_str: str | None) -> bool: def redact_job_bodies(yaml_str: str) -> str: - """Return workflow YAML with job bodies replaced by a placeholder.""" + """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]" @@ -107,6 +114,17 @@ def redact_job_bodies(yaml_str: str) -> str: 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. From 6e2fff7faddff70cbad9568a122585c9ec5909fa Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Thu, 16 Jul 2026 20:54:02 +0100 Subject: [PATCH 03/11] udpate subagent prompt --- services/job_chat/prompt.py | 31 ++++++++++++++----------------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/services/job_chat/prompt.py b/services/job_chat/prompt.py index 4092c154..a252f954 100644 --- a/services/job_chat/prompt.py +++ b/services/job_chat/prompt.py @@ -179,23 +179,20 @@ # Appended in subagent mode only (when job_chat is called from global_chat). subagent_mode_instructions = """ -NEVER say you cannot see the workflow, a step, or its code; never mention your -tools or access; never send the user to another page. - -`edit_job` edits THIS step only. Changes to anything else — workflow structure -(add/remove/rename steps, triggers, edges, adaptors) or another step's code — -go through `inspect_workflow`: call it as your very first action, with no -reply needed (at most: "I'll take a look at your workflow."). - -If the user mentions code that is not in (a warning, function, or -behavior you can't find — even if they say "this step"), they might be describing -another step. Never reply that it isn't in this step: to -change it, call `inspect_workflow`; to answer a question about it, read it -with `inspect_job_code`. - -Use `inspect_job_code` to read other steps whenever it helps — e.g. to see -what an upstream step outputs, or to keep style, patterns, or field names -consistent across steps. +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 `inspect_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. """ From 95ca77f1b32d65bd47ecca9e452a7aaac84c7e6c Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Wed, 22 Jul 2026 09:38:50 +0100 Subject: [PATCH 04/11] add navigation info --- services/global_chat/planner.py | 2 +- services/global_chat/router.py | 15 ++- .../job_code/test_canvas_code_request.md | 112 +++++++++++++++++ .../test_question_reading_another_step.md | 114 ++++++++++++++++++ services/job_chat/prompt.py | 50 ++++++-- .../tests/unit/test_subagent_prompt.py | 36 +++++- services/yaml_utils.py | 39 ++++-- 7 files changed, 340 insertions(+), 28 deletions(-) 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 diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index 45a7b472..38cfaba1 100644 --- a/services/global_chat/planner.py +++ b/services/global_chat/planner.py @@ -284,7 +284,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})" diff --git a/services/global_chat/router.py b/services/global_chat/router.py index 771b936b..f74e4191 100644 --- a/services/global_chat/router.py +++ b/services/global_chat/router.py @@ -22,7 +22,7 @@ from streaming_util import StreamManager from global_chat.config_loader import ConfigLoader from models import resolve_model -from 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__) @@ -357,6 +357,19 @@ def _route_to_job_chat( # 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) 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/job_chat/prompt.py b/services/job_chat/prompt.py index a252f954..9f7eb80a 100644 --- a/services/job_chat/prompt.py +++ b/services/job_chat/prompt.py @@ -4,7 +4,7 @@ import sentry_sdk from langfuse import observe from util import create_logger, ApolloError, AdaptorSpecifier, get_db_connection -from yaml_utils import redact_job_bodies +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 @@ -292,6 +292,32 @@ def has(self, key): return hasattr(self, key) and getattr(self, key) is not 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 {})) @@ -388,20 +414,22 @@ def generate_system_message(context_dict, search_results, download_adaptor_docs= 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) - if focused: - focused_line = f"The focused step — the only one whose code you can edit — is '{focused}'." - else: - focused_line = ( - "No step is focused, so code edits cannot be applied here — " - "for code changes call `inspect_workflow`." - ) + 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"\nThe full workflow, job code redacted. {focused_line} " - f"READ other steps' code with `inspect_job_code` when the request refers to them.\n\n" - f"{redacted}\n" + f"\n{' '.join(header)}\n\n{redacted}\n" ) # Output contract goes LAST so it is the final, most prominent instruction. diff --git a/services/job_chat/tests/unit/test_subagent_prompt.py b/services/job_chat/tests/unit/test_subagent_prompt.py index 2dcf9d6a..b007241c 100644 --- a/services/job_chat/tests/unit/test_subagent_prompt.py +++ b/services/job_chat/tests/unit/test_subagent_prompt.py @@ -41,13 +41,37 @@ def test_subagent_prompt_strips_navigate_instruction(): assert "" in text -def test_subagent_prompt_names_focused_step(): - text = system_text(subagent=True, workflow_yaml=WORKFLOW_YAML) - assert "No step is focused" in text - +def _subagent_text(context: dict) -> str: blocks = generate_system_message( - context_dict={"job_key": "fetch-patients"}, search_results=None, + context_dict=context, search_results=None, subagent=True, workflow_yaml=WORKFLOW_YAML, ) - text = "\n".join(b["text"] for b in blocks) + 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/yaml_utils.py b/services/yaml_utils.py index b993ecc3..17920ec7 100644 --- a/services/yaml_utils.py +++ b/services/yaml_utils.py @@ -9,23 +9,44 @@ 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 step name from page URL. + 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 """ - 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 + view, step = get_page_view(page) + return step if view == "step" else None def normalize_name(name: str) -> str: From 982d947a1e6f01147d354fc609841dbe1f66ed02 Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Wed, 22 Jul 2026 16:59:34 +0100 Subject: [PATCH 05/11] add langfuse tracing --- services/job_chat/job_chat.py | 20 ++++++++++++++++---- services/workflow_chat/workflow_chat.py | 6 ++++++ 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/services/job_chat/job_chat.py b/services/job_chat/job_chat.py index 5bd929b9..dd509da2 100644 --- a/services/job_chat/job_chat.py +++ b/services/job_chat/job_chat.py @@ -92,9 +92,8 @@ "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, " - "with no reply text before it. To merely READ another step's code, use " - "inspect_job_code instead." + "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", @@ -384,6 +383,11 @@ def generate( 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 == "inspect_workflow"), None) if handover_block: @@ -401,7 +405,9 @@ def generate( tool_results = [] for block in tool_uses: if block.name == "inspect_job_code": - result_text = inspect_job_code(workflow_yaml, (block.input or {}).get("job_keys") or []) + 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 " @@ -809,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"), diff --git a/services/workflow_chat/workflow_chat.py b/services/workflow_chat/workflow_chat.py index 3eb8f890..67ee6711 100644 --- a/services/workflow_chat/workflow_chat.py +++ b/services/workflow_chat/workflow_chat.py @@ -762,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, From bb57c064fe80e78b873ac29b826d4232f278d679 Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Wed, 22 Jul 2026 19:21:11 +0100 Subject: [PATCH 06/11] rename inspect_workflow tool to edit_workflow --- services/job_chat/job_chat.py | 10 +++++----- services/job_chat/prompt.py | 6 +++--- services/job_chat/tests/unit/test_subagent_prompt.py | 4 ++-- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/services/job_chat/job_chat.py b/services/job_chat/job_chat.py index dd509da2..368f8573 100644 --- a/services/job_chat/job_chat.py +++ b/services/job_chat/job_chat.py @@ -87,8 +87,8 @@ # 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. -_INSPECT_WORKFLOW_TOOL = { - "name": "inspect_workflow", +_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) " @@ -117,7 +117,7 @@ "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 inspect_workflow " + "shape it produces. To change another step's code, call edit_workflow " "instead. Pass all the job keys you need in a single call." ), } @@ -297,7 +297,7 @@ def generate( if suggest_code: tools.append(_EDIT_TOOL) if subagent: - tools.append(_INSPECT_WORKFLOW_TOOL) + 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 {} @@ -389,7 +389,7 @@ def generate( round_index, ", ".join(b.name for b in tool_uses), ) - handover_block = next((b for b in tool_uses if b.name == "inspect_workflow"), None) + 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 diff --git a/services/job_chat/prompt.py b/services/job_chat/prompt.py index 9f7eb80a..fb1afc11 100644 --- a/services/job_chat/prompt.py +++ b/services/job_chat/prompt.py @@ -36,7 +36,7 @@ """ # Production only. In global chat's subagent mode, structure requests are -# handled via inspect_workflow, so this scope restriction is omitted entirely +# 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. @@ -188,7 +188,7 @@ 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 `inspect_workflow` as your very first action, before any + "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 @@ -492,7 +492,7 @@ def build_prompt(content, history, context, rag=None, api_key=None, stream_manag 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 `inspect_workflow` first;" + " 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({ diff --git a/services/job_chat/tests/unit/test_subagent_prompt.py b/services/job_chat/tests/unit/test_subagent_prompt.py index b007241c..a75fc808 100644 --- a/services/job_chat/tests/unit/test_subagent_prompt.py +++ b/services/job_chat/tests/unit/test_subagent_prompt.py @@ -24,7 +24,7 @@ def test_production_prompt_keeps_navigate_instruction(): # 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 "inspect_workflow" not in text + assert "edit_workflow" not in text assert "" not in text @@ -37,7 +37,7 @@ def test_subagent_prompt_strips_navigate_instruction(): 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 "inspect_workflow" in text + assert "edit_workflow" in text assert "" in text From 91a2f0b921cbdf8611ef3d8dc8cf65d3f3619f40 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Wed, 29 Jul 2026 11:31:25 +0100 Subject: [PATCH 07/11] changeset --- .changeset/five-bags-rule.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .changeset/five-bags-rule.md 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 From 788ce0871174bb98f93ea7828998588b84c3fedf Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Wed, 29 Jul 2026 11:37:52 +0100 Subject: [PATCH 08/11] resolve conflict --- services/global_chat/planner.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index 38cfaba1..9e603ac1 100644 --- a/services/global_chat/planner.py +++ b/services/global_chat/planner.py @@ -100,15 +100,12 @@ def run( """ logger.info("Planner.run() called") -<<<<<<< HEAD -======= 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) ->>>>>>> d8fec45 (add subagent mode and inspect tool) self.current_yaml = workflow_yaml self.yaml_modified = False self._user = user From 4797cf22c7417e1e33e4af525da51c353b61ba6b Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Wed, 29 Jul 2026 11:58:28 +0100 Subject: [PATCH 09/11] update test fixtures --- services/global_chat/tests/unit/test_router.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/services/global_chat/tests/unit/test_router.py b/services/global_chat/tests/unit/test_router.py index ca514ec9..8f97c499 100644 --- a/services/global_chat/tests/unit/test_router.py +++ b/services/global_chat/tests/unit/test_router.py @@ -108,6 +108,7 @@ def test_job_route_sends_subagent_payload() -> None: def make_planner_result() -> RouterResult: return RouterResult( response="planner answer", + response_segments=[{"type": "text", "content": "planner answer"}], attachments=[], history=[], usage={"input_tokens": 10}, @@ -168,7 +169,14 @@ def test_low_confidence_direct_route_goes_to_planner() -> None: 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", attachments=[], history=[], usage={}, meta={}) + 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, \ From d35992eabf2d495a0129c43963ca290b7949bafc Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Wed, 22 Jul 2026 19:32:50 +0100 Subject: [PATCH 10/11] add coherence instruction --- services/global_chat/prompts.yaml | 18 +++- .../test_rest_to_rest_sync_with_cron.md | 92 +++++++++++++++++++ 2 files changed, 108 insertions(+), 2 deletions(-) create mode 100644 services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md diff --git a/services/global_chat/prompts.yaml b/services/global_chat/prompts.yaml index 4da24ee0..59559ae9 100644 --- a/services/global_chat/prompts.yaml +++ b/services/global_chat/prompts.yaml @@ -108,8 +108,22 @@ 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. Define between-step contracts (before dispatching job code). For any edge where a + step depends on an upstream step's output, decide 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. + + When two steps share data, 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." + 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/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. From 1a3f2a2dab2e62ecc03162b765d51cd5f7784aa9 Mon Sep 17 00:00:00 2001 From: hanna-paasivirta Date: Tue, 28 Jul 2026 18:11:03 +0100 Subject: [PATCH 11/11] fix prompt for existing workflows --- services/global_chat/prompts.yaml | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/services/global_chat/prompts.yaml b/services/global_chat/prompts.yaml index 59559ae9..10712118 100644 --- a/services/global_chat/prompts.yaml +++ b/services/global_chat/prompts.yaml @@ -108,20 +108,24 @@ 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. Define between-step contracts (before dispatching job code). For any edge where a - step depends on an upstream step's output, decide in system-agnostic terms: + 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. - When two steps share data, 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. + 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