Handle routing failures

.changeset/five-bags-rule.md

@@ -0,0 +1,6 @@
+---
+"apollo": patch
+---
+
+global_chat: enable subagents to pull missing context, recovering from routing
+errors

services/global_chat/planner.py

@@ -26,7 +26,7 @@
 from global_chat.config_loader import ConfigLoader
 from models import resolve_model
 from global_chat.tools.tool_definitions import TOOL_DEFINITIONS
-from global_chat.yaml_utils import stitch_job_code, redact_job_bodies, find_job_in_yaml, get_step_name_from_page
+from yaml_utils import stitch_job_code, redact_job_bodies, find_job_in_yaml, get_step_name_from_page, inspect_job_code
 from tools.search_documentation.search_documentation import search_documentation_tool
 from global_chat.subagent_caller import call_workflow_agent, call_job_agent, format_subagent_result_for_llm
 
@@ -81,6 +81,7 @@ def run(
         stream: bool,
         user: Optional[Dict] = None,
         metrics_opt_in: Optional[bool] = None,
+        stream_manager: Optional[StreamManager] = None,
     ) -> PlannerResult:
         """
         Run the planner agent with tool-calling loop.
@@ -91,12 +92,20 @@ def run(
             page: Current page URL (e.g. workflows/name/step-name)
             history: Conversation history
             stream: Whether to stream text via SSE events
+            stream_manager: Optional shared stream manager from the router, so
+                a handed-over request continues on the same stream
 
         Returns:
             PlannerResult with response, attachments, history, usage, meta
         """
         logger.info("Planner.run() called")
 
+        stream_manager = stream_manager or StreamManager(model=self.model, stream=stream)
+        if workflow_yaml:
+            stream_manager.send_thinking(STATUS_REVIEWING_WORKFLOW + STATUS_PLANNING)
+        else:
+            stream_manager.send_thinking(STATUS_NEW_WORKFLOW + STATUS_PLANNING)
+
         self.current_yaml = workflow_yaml
         self.yaml_modified = False
         self._user = user
@@ -272,7 +281,7 @@ def _build_user_content(self, content: str, page: Optional[str]) -> str:
                 matched_key, _ = find_job_in_yaml(self.current_yaml, step_name)
                 step_name = matched_key or step_name
             if step_name:
-                user_content += f"\n\n(The user is currently viewing the step '{step_name}' — \"this step\" refers to it.)"
+                user_content += f"\n\n(The user is currently viewing the step '{step_name}'.)"
             else:
                 user_content += f"\n\n(The user is currently viewing: {page})"
 
@@ -488,19 +497,7 @@ def _execute_tool(self, tool_use_block, stream_manager, total_usage, tool_calls_
             if single_key:
                 job_keys.append(single_key)
 
-            if not self.current_yaml:
-                tool_result = "No workflow available to inspect."
-            elif not job_keys:
-                tool_result = "ERROR: No job keys provided."
-            else:
-                parts = []
-                for job_key in job_keys:
-                    _, job_data = find_job_in_yaml(self.current_yaml, job_key)
-                    if job_data and job_data.get("body"):
-                        parts.append(f"Job code for '{job_key}':\n\n{job_data['body']}")
-                    else:
-                        parts.append(f"No code found for job '{job_key}'.")
-                tool_result = "\n\n".join(parts)
+            tool_result = inspect_job_code(self.current_yaml, job_keys)
 
             tool_calls_meta.append({"tool": "inspect_job_code", "input": tool_use_block.input})
 

services/global_chat/prompts.yaml

@@ -108,8 +108,26 @@ prompts:
     Job code is stitched into the workflow YAML by job_key — the workflow must exist first.
 
     1. Create/modify workflow structure FIRST (`call_workflow_agent`)
-    2. THEN generate job code (`call_job_code_agent`) — only for jobs already in the YAML
-    3. Set `job_key` to the exact key from the workflow structure
+    2. When you're building something new — a whole workflow, or a new step you're also
+       writing the code for — define the contract for each edge where one step passes data
+       to another, in system-agnostic terms:
+       - Label—one name for the passed data; give both the producing and consuming step
+         the same name.
+       - Ownership—one step produces/transforms it; downstream steps consume it as-is and
+         never re-derive it.
+
+       Put the identical contract line in both `call_job_code_agent` messages. Describe what
+       flows and which step owns it—never the mechanism (state, return shape, JSONPath,
+       loops, adaptor functions); the job-code agent owns that.
+
+       e.g. "Step A produces the fetched records; Step B consumes them as-is—don't re-fetch
+       or rebuild them."
+
+       This only applies to work you are creating. When editing existing steps, make only
+       the change the user asked for—don't restate contracts, re-derive the data flow, or
+       make other changes they didn't request.
+    3. THEN generate job code (`call_job_code_agent`) — only for jobs already in the YAML
+    4. Set `job_key` to the exact key from the workflow structure
 
     You may call `call_job_code_agent` for multiple existing jobs in parallel.
     Never call `call_job_code_agent` and `call_workflow_agent` in the same step.

services/global_chat/router.py

@@ -17,11 +17,12 @@
 
 sys.path.append(str(Path(__file__).parent.parent))
 
-from langfuse import observe
+from langfuse import observe, get_client as get_langfuse_client
 from util import create_logger, ApolloError, sum_usage
+from streaming_util import StreamManager
 from global_chat.config_loader import ConfigLoader
 from models import resolve_model
-from global_chat.yaml_utils import get_step_name_from_page, find_job_in_yaml, stitch_job_code, workflow_has_job_code
+from yaml_utils import get_step_name_from_page, get_page_view, find_job_in_yaml, stitch_job_code, workflow_has_job_code
 
 logger = create_logger(__name__)
 
@@ -111,6 +112,10 @@ def route_and_execute(
         self._input_attachments = attachments or []
         self._user = user
         self._metrics_opt_in = metrics_opt_in
+        # One stream manager shared by whichever agents serve this request, so
+        # a handed-over request continues the same stream instead of starting
+        # a second message lifecycle.
+        self._stream_manager = StreamManager(model=self.model, stream=stream)
 
         try:
             decision = self._make_routing_decision(content, workflow_yaml, page, history)
@@ -121,8 +126,18 @@ def route_and_execute(
             logger.warning(f"Routing decision failed: {e}. Defaulting to planner for safety.")
             decision = RouterDecision(destination="planner", confidence=1)
 
+        # Direct routes are a fast path for clear-cut requests; when the router
+        # itself is unsure, take the path that can't be wrong. Costs nothing:
+        # the confidence comes back in the same routing call.
+        if decision.destination in ("workflow_agent", "job_code_agent") and decision.confidence < 3:
+            logger.warning(
+                f"Low router confidence ({decision.confidence}) for {decision.destination} — routing to planner instead"
+            )
+            self._track_reroute({"low_confidence_reroute": decision.destination})
+            decision = RouterDecision(destination="planner", confidence=decision.confidence)
+
         if decision.destination == "workflow_agent":
-            result = self._route_to_workflow_chat(content, workflow_yaml, history, stream, decision.confidence)
+            result = self._route_to_workflow_chat(content, workflow_yaml, page, history, stream, decision.confidence)
         elif decision.destination == "job_code_agent":
             result = self._route_to_job_chat(
                 content, workflow_yaml, page, history, stream, decision.confidence, decision.job_key
@@ -229,7 +244,7 @@ def _format_attachments_for_content(self, content: str) -> str:
         return "\n".join(parts)
 
     def _route_to_workflow_chat(
-        self, content: str, workflow_yaml: Optional[str], history: List[Dict], stream: bool, confidence: int
+        self, content: str, workflow_yaml: Optional[str], page: Optional[str], history: List[Dict], stream: bool, confidence: int
     ) -> RouterResult:
         """Route directly to workflow_chat."""
         from workflow_chat.workflow_chat import main as workflow_chat_main
@@ -247,9 +262,17 @@ def _route_to_workflow_chat(
             "api_key": self.api_key,
             "meta": {"user": self._user} if self._user else None,
             "metrics_opt_in": self._metrics_opt_in,
+            "subagent": True,
+            "_stream_manager": self._stream_manager,
         }
 
         result = workflow_chat_main(payload)
+
+        if result.get("handover"):
+            return self._handover_to_planner(
+                "workflow_agent", result, content, workflow_yaml, page, history, stream, confidence
+            )
+
         total_usage = sum_usage(self.routing_usage, result["usage"])
 
         attachments = []
@@ -330,6 +353,22 @@ def _route_to_job_chat(
                 job_context["adaptor"] = job_data["adaptor"]
             if job_data.get("name"):
                 job_context["page_name"] = job_data["name"]
+        if matched_job_key:
+            # Tells job_chat's subagent prompt which step is focused/editable
+            job_context["job_key"] = matched_job_key
+
+        # What the user actually has on screen, independent of which step we
+        # focus for editing: a specific step's code, or the workflow canvas.
+        # Only the router knows this (planner/prod calls omit it, so the prompt
+        # grounding line stays off). Fail safe: only claim a step the page name
+        # resolves to a real job — a mis-split name simply yields no line.
+        page_view, page_step = get_page_view(page)
+        if page_view == "step" and workflow_yaml:
+            _, viewed_job = find_job_in_yaml(workflow_yaml, page_step)
+            if viewed_job and viewed_job.get("name"):
+                job_context["viewing"] = viewed_job["name"]
+        elif page_view == "overview":
+            job_context["viewing"] = "canvas"
 
         clean_history = [{"role": t["role"], "content": t["content"]} for t in history]
         enriched_content = self._format_attachments_for_content(content)
@@ -343,9 +382,18 @@ def _route_to_job_chat(
             "api_key": self.api_key,
             "meta": {"user": self._user} if self._user else None,
             "metrics_opt_in": self._metrics_opt_in,
+            "subagent": True,
+            "workflow_yaml": workflow_yaml,
+            "_stream_manager": self._stream_manager,
         }
 
         result = job_chat_main(payload)
+
+        if result.get("handover"):
+            return self._handover_to_planner(
+                "job_code_agent", result, content, workflow_yaml, page, history, stream, confidence
+            )
+
         total_usage = sum_usage(self.routing_usage, result["usage"])
 
         # Stitch suggested_code back into the workflow YAML. The full YAML is
@@ -369,6 +417,46 @@ def _route_to_job_chat(
             meta={"agents": ["router", "job_code_agent"], "router_confidence": confidence},
         )
 
+    def _handover_to_planner(
+        self,
+        from_agent: str,
+        subagent_result: Dict,
+        content: str,
+        workflow_yaml: Optional[str],
+        page: Optional[str],
+        history: List[Dict],
+        stream: bool,
+        confidence: int,
+    ) -> RouterResult:
+        """Reroute a handed-over request to the planner.
+
+        A direct-routed subagent signalled it cannot complete the request
+        (wrong route or missing capability). The planner never hands over, so
+        this retries at most once. The shared stream manager means the user
+        never sees the aborted attempt.
+        """
+        reason = subagent_result["handover"]
+        logger.warning(f"{from_agent} handed over: {reason}. Rerouting to planner")
+        self._track_reroute({"handover_from": from_agent, "handover_reason": reason})
+
+        planner_result = self._route_to_planner(content, workflow_yaml, page, history, stream, confidence)
+        planner_result.usage = sum_usage(planner_result.usage, subagent_result.get("usage", {}))
+        return planner_result
+
+    def _track_reroute(self, metadata: Dict) -> None:
+        """Record reroute diagnostics on the Langfuse trace (opt-in per request).
+
+        Deliberately kept out of the response meta: the frontend does nothing
+        with these, they are for Langfuse analysis only. Server logs carry the
+        same information when tracking is off.
+        """
+        if not self._metrics_opt_in:
+            return
+        try:
+            get_langfuse_client().update_current_span(metadata=metadata)
+        except Exception:
+            logger.warning("Failed to record reroute metadata in Langfuse")
+
     def _route_to_planner(
         self,
         content: str,
@@ -395,6 +483,7 @@ def _route_to_planner(
             stream=stream,
             user=self._user,
             metrics_opt_in=self._metrics_opt_in,
+            stream_manager=self._stream_manager,
         )
 
         total_usage = sum_usage(self.routing_usage, planner_result.usage)

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__)
 

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.