diff --git a/.changeset/five-bags-rule.md b/.changeset/five-bags-rule.md new file mode 100644 index 00000000..d8c883b6 --- /dev/null +++ b/.changeset/five-bags-rule.md @@ -0,0 +1,6 @@ +--- +"apollo": patch +--- + +global_chat: enable subagents to pull missing context, recovering from routing +errors diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index 5fec69ff..9e603ac1 100644 --- a/services/global_chat/planner.py +++ b/services/global_chat/planner.py @@ -26,7 +26,7 @@ from global_chat.config_loader import ConfigLoader from models import resolve_model from global_chat.tools.tool_definitions import TOOL_DEFINITIONS -from global_chat.yaml_utils import stitch_job_code, redact_job_bodies, find_job_in_yaml, get_step_name_from_page +from yaml_utils import stitch_job_code, redact_job_bodies, find_job_in_yaml, get_step_name_from_page, inspect_job_code from tools.search_documentation.search_documentation import search_documentation_tool from global_chat.subagent_caller import call_workflow_agent, call_job_agent, format_subagent_result_for_llm @@ -81,6 +81,7 @@ def run( stream: bool, user: Optional[Dict] = None, metrics_opt_in: Optional[bool] = None, + stream_manager: Optional[StreamManager] = None, ) -> PlannerResult: """ Run the planner agent with tool-calling loop. @@ -91,12 +92,20 @@ def run( page: Current page URL (e.g. workflows/name/step-name) history: Conversation history stream: Whether to stream text via SSE events + stream_manager: Optional shared stream manager from the router, so + a handed-over request continues on the same stream Returns: PlannerResult with response, attachments, history, usage, meta """ logger.info("Planner.run() called") + stream_manager = stream_manager or StreamManager(model=self.model, stream=stream) + if workflow_yaml: + stream_manager.send_thinking(STATUS_REVIEWING_WORKFLOW + STATUS_PLANNING) + else: + stream_manager.send_thinking(STATUS_NEW_WORKFLOW + STATUS_PLANNING) + self.current_yaml = workflow_yaml self.yaml_modified = False self._user = user @@ -272,7 +281,7 @@ def _build_user_content(self, content: str, page: Optional[str]) -> str: matched_key, _ = find_job_in_yaml(self.current_yaml, step_name) step_name = matched_key or step_name if step_name: - user_content += f"\n\n(The user is currently viewing the step '{step_name}' — \"this step\" refers to it.)" + user_content += f"\n\n(The user is currently viewing the step '{step_name}'.)" else: user_content += f"\n\n(The user is currently viewing: {page})" @@ -488,19 +497,7 @@ def _execute_tool(self, tool_use_block, stream_manager, total_usage, tool_calls_ if single_key: job_keys.append(single_key) - if not self.current_yaml: - tool_result = "No workflow available to inspect." - elif not job_keys: - tool_result = "ERROR: No job keys provided." - else: - parts = [] - for job_key in job_keys: - _, job_data = find_job_in_yaml(self.current_yaml, job_key) - if job_data and job_data.get("body"): - parts.append(f"Job code for '{job_key}':\n\n{job_data['body']}") - else: - parts.append(f"No code found for job '{job_key}'.") - tool_result = "\n\n".join(parts) + tool_result = inspect_job_code(self.current_yaml, job_keys) tool_calls_meta.append({"tool": "inspect_job_code", "input": tool_use_block.input}) diff --git a/services/global_chat/prompts.yaml b/services/global_chat/prompts.yaml index 4da24ee0..10712118 100644 --- a/services/global_chat/prompts.yaml +++ b/services/global_chat/prompts.yaml @@ -108,8 +108,26 @@ prompts: Job code is stitched into the workflow YAML by job_key — the workflow must exist first. 1. Create/modify workflow structure FIRST (`call_workflow_agent`) - 2. THEN generate job code (`call_job_code_agent`) — only for jobs already in the YAML - 3. Set `job_key` to the exact key from the workflow structure + 2. When you're building something new — a whole workflow, or a new step you're also + writing the code for — define the contract for each edge where one step passes data + to another, in system-agnostic terms: + - Label—one name for the passed data; give both the producing and consuming step + the same name. + - Ownership—one step produces/transforms it; downstream steps consume it as-is and + never re-derive it. + + Put the identical contract line in both `call_job_code_agent` messages. Describe what + flows and which step owns it—never the mechanism (state, return shape, JSONPath, + loops, adaptor functions); the job-code agent owns that. + + e.g. "Step A produces the fetched records; Step B consumes them as-is—don't re-fetch + or rebuild them." + + This only applies to work you are creating. When editing existing steps, make only + the change the user asked for—don't restate contracts, re-derive the data flow, or + make other changes they didn't request. + 3. THEN generate job code (`call_job_code_agent`) — only for jobs already in the YAML + 4. Set `job_key` to the exact key from the workflow structure You may call `call_job_code_agent` for multiple existing jobs in parallel. Never call `call_job_code_agent` and `call_workflow_agent` in the same step. diff --git a/services/global_chat/router.py b/services/global_chat/router.py index db0bdcf3..f74e4191 100644 --- a/services/global_chat/router.py +++ b/services/global_chat/router.py @@ -17,11 +17,12 @@ sys.path.append(str(Path(__file__).parent.parent)) -from langfuse import observe +from langfuse import observe, get_client as get_langfuse_client from util import create_logger, ApolloError, sum_usage +from streaming_util import StreamManager from global_chat.config_loader import ConfigLoader from models import resolve_model -from global_chat.yaml_utils import get_step_name_from_page, find_job_in_yaml, stitch_job_code, workflow_has_job_code +from yaml_utils import get_step_name_from_page, get_page_view, find_job_in_yaml, stitch_job_code, workflow_has_job_code logger = create_logger(__name__) @@ -111,6 +112,10 @@ def route_and_execute( self._input_attachments = attachments or [] self._user = user self._metrics_opt_in = metrics_opt_in + # One stream manager shared by whichever agents serve this request, so + # a handed-over request continues the same stream instead of starting + # a second message lifecycle. + self._stream_manager = StreamManager(model=self.model, stream=stream) try: decision = self._make_routing_decision(content, workflow_yaml, page, history) @@ -121,8 +126,18 @@ def route_and_execute( logger.warning(f"Routing decision failed: {e}. Defaulting to planner for safety.") decision = RouterDecision(destination="planner", confidence=1) + # Direct routes are a fast path for clear-cut requests; when the router + # itself is unsure, take the path that can't be wrong. Costs nothing: + # the confidence comes back in the same routing call. + if decision.destination in ("workflow_agent", "job_code_agent") and decision.confidence < 3: + logger.warning( + f"Low router confidence ({decision.confidence}) for {decision.destination} — routing to planner instead" + ) + self._track_reroute({"low_confidence_reroute": decision.destination}) + decision = RouterDecision(destination="planner", confidence=decision.confidence) + if decision.destination == "workflow_agent": - result = self._route_to_workflow_chat(content, workflow_yaml, history, stream, decision.confidence) + result = self._route_to_workflow_chat(content, workflow_yaml, page, history, stream, decision.confidence) elif decision.destination == "job_code_agent": result = self._route_to_job_chat( content, workflow_yaml, page, history, stream, decision.confidence, decision.job_key @@ -229,7 +244,7 @@ def _format_attachments_for_content(self, content: str) -> str: return "\n".join(parts) def _route_to_workflow_chat( - self, content: str, workflow_yaml: Optional[str], history: List[Dict], stream: bool, confidence: int + self, content: str, workflow_yaml: Optional[str], page: Optional[str], history: List[Dict], stream: bool, confidence: int ) -> RouterResult: """Route directly to workflow_chat.""" from workflow_chat.workflow_chat import main as workflow_chat_main @@ -247,9 +262,17 @@ def _route_to_workflow_chat( "api_key": self.api_key, "meta": {"user": self._user} if self._user else None, "metrics_opt_in": self._metrics_opt_in, + "subagent": True, + "_stream_manager": self._stream_manager, } result = workflow_chat_main(payload) + + if result.get("handover"): + return self._handover_to_planner( + "workflow_agent", result, content, workflow_yaml, page, history, stream, confidence + ) + total_usage = sum_usage(self.routing_usage, result["usage"]) attachments = [] @@ -330,6 +353,22 @@ def _route_to_job_chat( job_context["adaptor"] = job_data["adaptor"] if job_data.get("name"): job_context["page_name"] = job_data["name"] + if matched_job_key: + # Tells job_chat's subagent prompt which step is focused/editable + job_context["job_key"] = matched_job_key + + # What the user actually has on screen, independent of which step we + # focus for editing: a specific step's code, or the workflow canvas. + # Only the router knows this (planner/prod calls omit it, so the prompt + # grounding line stays off). Fail safe: only claim a step the page name + # resolves to a real job — a mis-split name simply yields no line. + page_view, page_step = get_page_view(page) + if page_view == "step" and workflow_yaml: + _, viewed_job = find_job_in_yaml(workflow_yaml, page_step) + if viewed_job and viewed_job.get("name"): + job_context["viewing"] = viewed_job["name"] + elif page_view == "overview": + job_context["viewing"] = "canvas" clean_history = [{"role": t["role"], "content": t["content"]} for t in history] enriched_content = self._format_attachments_for_content(content) @@ -343,9 +382,18 @@ def _route_to_job_chat( "api_key": self.api_key, "meta": {"user": self._user} if self._user else None, "metrics_opt_in": self._metrics_opt_in, + "subagent": True, + "workflow_yaml": workflow_yaml, + "_stream_manager": self._stream_manager, } result = job_chat_main(payload) + + if result.get("handover"): + return self._handover_to_planner( + "job_code_agent", result, content, workflow_yaml, page, history, stream, confidence + ) + total_usage = sum_usage(self.routing_usage, result["usage"]) # Stitch suggested_code back into the workflow YAML. The full YAML is @@ -369,6 +417,46 @@ def _route_to_job_chat( meta={"agents": ["router", "job_code_agent"], "router_confidence": confidence}, ) + def _handover_to_planner( + self, + from_agent: str, + subagent_result: Dict, + content: str, + workflow_yaml: Optional[str], + page: Optional[str], + history: List[Dict], + stream: bool, + confidence: int, + ) -> RouterResult: + """Reroute a handed-over request to the planner. + + A direct-routed subagent signalled it cannot complete the request + (wrong route or missing capability). The planner never hands over, so + this retries at most once. The shared stream manager means the user + never sees the aborted attempt. + """ + reason = subagent_result["handover"] + logger.warning(f"{from_agent} handed over: {reason}. Rerouting to planner") + self._track_reroute({"handover_from": from_agent, "handover_reason": reason}) + + planner_result = self._route_to_planner(content, workflow_yaml, page, history, stream, confidence) + planner_result.usage = sum_usage(planner_result.usage, subagent_result.get("usage", {})) + return planner_result + + def _track_reroute(self, metadata: Dict) -> None: + """Record reroute diagnostics on the Langfuse trace (opt-in per request). + + Deliberately kept out of the response meta: the frontend does nothing + with these, they are for Langfuse analysis only. Server logs carry the + same information when tracking is off. + """ + if not self._metrics_opt_in: + return + try: + get_langfuse_client().update_current_span(metadata=metadata) + except Exception: + logger.warning("Failed to record reroute metadata in Langfuse") + def _route_to_planner( self, content: str, @@ -395,6 +483,7 @@ def _route_to_planner( stream=stream, user=self._user, metrics_opt_in=self._metrics_opt_in, + stream_manager=self._stream_manager, ) total_usage = sum_usage(self.routing_usage, planner_result.usage) diff --git a/services/global_chat/subagent_caller.py b/services/global_chat/subagent_caller.py index 19d90653..6c34ce1b 100644 --- a/services/global_chat/subagent_caller.py +++ b/services/global_chat/subagent_caller.py @@ -13,7 +13,7 @@ from langfuse import observe from util import create_logger, ApolloError -from global_chat.yaml_utils import find_job_in_yaml +from yaml_utils import find_job_in_yaml logger = create_logger(__name__) diff --git a/services/global_chat/tests/acceptance/job_code/test_canvas_code_request.md b/services/global_chat/tests/acceptance/job_code/test_canvas_code_request.md new file mode 100644 index 00000000..31f42445 --- /dev/null +++ b/services/global_chat/tests/acceptance/job_code/test_canvas_code_request.md @@ -0,0 +1,112 @@ +--- +id: global-chat.job-code.canvas-code-request +service: global_chat +judges: [general, openfn_code_quality] +--- + +# notes + +The user is on the workflow canvas (a 2-segment page URL, no step open) but asks +for a code change to a named step. The router should resolve this to +job_code_agent for the fetch-orders step, so job_chat is told the user is viewing +the canvas while fetch-orders is the step it can edit — the case where the +on-screen view and the editable step deliberately differ. + +Watch for two things. First, the edit must land on the fetch-orders step (not be +refused because "no step is open", and not bodged elsewhere). Second, the reply +must read like a normal answer: it must not surface internal mechanics (routing, +agents, subagents) or treat being on the canvas / not having a step open as a +limitation it narrates to the user. + +# quality_criteria + +- The fetch-orders step is updated to log a warning (e.g. a console.warn) when the API returns no orders — a guard that checks the fetched orders are empty. +- Only the fetch-orders step is changed; the other steps are left unchanged. +- The reply reads as a direct answer to the request and does NOT mention internal mechanics (routing, agents, subagents) or frame "being on the canvas" / "no step open" as a reason it cannot help. + +# settings + +## page + +workflows/orders-sync + +## workflow_yaml + +```yaml +name: orders-sync +jobs: + fetch-orders: + id: job-fetch-orders-id + name: Fetch Orders + adaptor: "@openfn/language-http@6.5.4" + body: | + get('/orders', { query: { since: $.lastRunAt } }); + fn(state => { + const orders = state.data.orders || []; + return { ...state, orders }; + }); + normalize-orders: + id: job-normalize-orders-id + name: Normalize Orders + adaptor: "@openfn/language-common@2.3.0" + body: | + fn(state => { + const orders = state.orders.map(o => ({ + id: o.id, + total: Number(o.total_price), + customerEmail: o.customer?.email, + placedAt: o.created_at + })); + return { ...state, orders }; + }); + notify-fulfillment: + id: job-notify-fulfillment-id + name: Notify Fulfillment + adaptor: "@openfn/language-http@6.5.4" + body: | + each( + $.orders, + post('https://fulfillment.example.org/queue', state => ({ + body: state.data + })) + ); +triggers: + cron: + id: trigger-cron-id + type: cron + cron_expression: "*/30 * * * *" + enabled: true +edges: + cron->fetch-orders: + id: edge-cron-fetch + source_trigger: cron + target_job: fetch-orders + condition_type: always + enabled: true + fetch-orders->normalize-orders: + id: edge-fetch-normalize + source_job: fetch-orders + target_job: normalize-orders + condition_type: on_job_success + enabled: true + normalize-orders->notify-fulfillment: + id: edge-normalize-notify + source_job: normalize-orders + target_job: notify-fulfillment + condition_type: on_job_success + enabled: true +``` + +## meta.session_id + +sess-job-code-canvas-code-request-0001 + +# turn + +## role + +user + +## content + +In the fetch-orders step, add a check that logs a warning if the API comes back with no orders. diff --git a/services/global_chat/tests/acceptance/job_code/test_question_reading_another_step.md b/services/global_chat/tests/acceptance/job_code/test_question_reading_another_step.md new file mode 100644 index 00000000..65c8be25 --- /dev/null +++ b/services/global_chat/tests/acceptance/job_code/test_question_reading_another_step.md @@ -0,0 +1,114 @@ +--- +id: global-chat.job-code.question-reading-another-step +service: global_chat +judges: [general, openfn_code_quality] +--- + +# notes + +The user is on the last step (notify-fulfillment) and asks what fields each order +has "at this point". The focused step only consumes `$.orders`; the shape those +orders actually have is defined by the UPSTREAM normalize-orders step, not by any +code visible in the focused step. To answer correctly the assistant has to read +the normalize-orders step and describe the fields it produces. + +This exercises the read-only path new to job_chat in subagent mode: it should +route to job_code_agent, use inspect_job_code to read the upstream step, and +answer — without escalating to the planner and without a code change. (The +planner could also field it; either way the answer must reflect the normalize +step's real output, not a generic guess.) The key failure mode to catch is the +model answering from thin air, or replying that it cannot see the data / the +other step. + +# quality_criteria + +- The response describes the normalized order shape produced upstream by the normalize-orders step: an id, a numeric total, a customerEmail, and a placedAt (timestamp). +- The answer is grounded in the actual upstream code, not a generic description of "an order", and it does NOT claim it cannot see the data or the other step. +- The response does NOT propose or apply a code change — the user asked a question. + +# settings + +## page + +workflows/orders-sync/notify-fulfillment + +## workflow_yaml + +```yaml +name: orders-sync +jobs: + fetch-orders: + id: job-fetch-orders-id + name: Fetch Orders + adaptor: "@openfn/language-http@6.5.4" + body: | + get('/orders', { query: { since: $.lastRunAt } }); + fn(state => { + const orders = state.data.orders || []; + return { ...state, orders }; + }); + normalize-orders: + id: job-normalize-orders-id + name: Normalize Orders + adaptor: "@openfn/language-common@2.3.0" + body: | + fn(state => { + const orders = state.orders.map(o => ({ + id: o.id, + total: Number(o.total_price), + customerEmail: o.customer?.email, + placedAt: o.created_at + })); + return { ...state, orders }; + }); + notify-fulfillment: + id: job-notify-fulfillment-id + name: Notify Fulfillment + adaptor: "@openfn/language-http@6.5.4" + body: | + each( + $.orders, + post('https://fulfillment.example.org/queue', state => ({ + body: state.data + })) + ); +triggers: + cron: + id: trigger-cron-id + type: cron + cron_expression: "*/30 * * * *" + enabled: true +edges: + cron->fetch-orders: + id: edge-cron-fetch + source_trigger: cron + target_job: fetch-orders + condition_type: always + enabled: true + fetch-orders->normalize-orders: + id: edge-fetch-normalize + source_job: fetch-orders + target_job: normalize-orders + condition_type: on_job_success + enabled: true + normalize-orders->notify-fulfillment: + id: edge-normalize-notify + source_job: normalize-orders + target_job: notify-fulfillment + condition_type: on_job_success + enabled: true +``` + +## meta.session_id + +sess-job-code-question-reading-another-step-0001 + +# turn + +## role + +user + +## content + +Before I post these to fulfillment, what fields does each order actually have at this point? diff --git a/services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md b/services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md new file mode 100644 index 00000000..a973260e --- /dev/null +++ b/services/global_chat/tests/acceptance/one_shot_workflows/test_rest_to_rest_sync_with_cron.md @@ -0,0 +1,92 @@ +--- +id: global-chat.rest-to-rest-sync-with-cron +service: global_chat +judges: [general, openfn_workflow_expert, openfn_code_quality] +--- + +# notes + +From-scratch scheduled REST-to-REST sync with fully specified job code. No existing YAML, no history. The user gives a precise spec: a daily cron trigger, an HTTP GET of a user list, a transform into a three-field shape (userId, title, body), and an HTTP POST of each transformed record. The planner should be invoked, call the workflow agent to produce the structure with a cron trigger, then call the job code agent to fill in the bodies. + +The key thing this test probes is data-flow coherence across steps: the transform step and the post step must agree on how the transformed records are passed between them. The steps should not read as if written in isolation — the downstream step must consume exactly what the upstream step produced, under the same name, without re-fetching or rebuilding it. + +The following workflow is a NON-BINDING reference showing one acceptable shape. Do not require the candidate to match it (adaptor versions, job names, whether GET and transform are one step or two, and the exact state key may all differ). Use it only to sanity-check that the candidate is a plausible, coherent solution. + +```yaml +name: " Daily REST Endpoint Sync (Manual)" +jobs: + Fetch-and-transform-users: + name: Fetch and transform users + adaptor: "@openfn/language-http@latest" + body: >- + + + get('https://jsonplaceholder.typicode.com/users'); + + + fn((state) => { + const users = state.data || []; + const records = users.map((user) => ({ + userId: user.id, + title: user.name, + body: `Email: ${user.email} | Company: ${user.company?.name ?? 'N/A'}`, + })); + console.log(`Transformed ${records.length} users`); + return { ...state, records }; + }); + Post-records-to-target: + name: Post records to target + adaptor: "@openfn/language-http@7.3.2" + body: > + each( + '$.records[*]', + post('https://jsonplaceholder.typicode.com/posts', (state) => state.data) + ); + + + fn((state) => { + console.log(`Posted ${state.records?.length ?? 0} records`); + return state; + }); +triggers: + cron: + type: cron + enabled: false + cron_expression: 0 0 * * * + cron_cursor_job: null +edges: + cron->Fetch-and-transform-users: + condition_type: always + enabled: true + target_job: Fetch-and-transform-users + source_trigger: cron + Fetch-and-transform-users->Post-records-to-target: + condition_type: on_job_success + enabled: true + target_job: Post-records-to-target + source_job: Fetch-and-transform-users +``` + +# quality_criteria + +- The workflow uses a cron trigger scheduled to run once a day (e.g. a `0 0 * * *` daily expression), not a webhook or a different frequency. +- A step fetches the user list from the source endpoint (`https://jsonplaceholder.typicode.com/users`) using an HTTP get. +- A transform maps each user into an object with exactly the three requested fields: `userId` (the user's id), `title` (the user's name), and `body` (a string combining the user's email and company name). +- A step POSTs each transformed record to the target endpoint (`https://jsonplaceholder.typicode.com/posts`) using an HTTP post. +- Data-flow coherence: the posting step consumes the exact data the transform step produced, referencing it under the same state key/name that the transform step wrote to. There is no key mismatch between the producing and consuming steps. +- The posting step does not re-fetch the users or rebuild the transformed objects itself — it consumes the upstream output as-is rather than duplicating the transform. +- The solution stays simple as requested: no branching, filtering, deduplication, or auth logic beyond what the user asked for. + +# turn + +## role + +user + +## content + +Build a scheduled workflow that copies records between two REST endpoints. +Trigger: cron, once a day. +Steps: GET the list of users from https://jsonplaceholder.typicode.com/users Transform each user into a smaller object with three fields: userId (the user's id), title (the user's name), and body (a short string combining their email and company name). POST each transformed record to https://jsonplaceholder.typicode.com/posts + +No authentication is required for this API. Keep it simple: no branching or deduplication. diff --git a/services/global_chat/tests/integration/__init__.py b/services/global_chat/tests/integration/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/services/global_chat/tests/integration/test_handover.py b/services/global_chat/tests/integration/test_handover.py new file mode 100644 index 00000000..f17521d2 --- /dev/null +++ b/services/global_chat/tests/integration/test_handover.py @@ -0,0 +1,136 @@ +"""Integration tests for subagent handover: curated misroutes sent directly to +job_chat and workflow_chat, as if the router had picked the wrong destination. + +These hit the live Anthropic API (manual/nightly, costs tokens). Each service +gets two scenarios: an obvious misroute, and an oblique one where the request +never names the other step — the model has to work out that what's being asked +lies beyond what it can see or edit. +""" + +import pytest +from dotenv import load_dotenv + +load_dotenv() + +from job_chat.job_chat import main as job_chat_main # noqa: E402 +from workflow_chat.workflow_chat import main as workflow_chat_main # noqa: E402 + +pytestmark = pytest.mark.integration + +WORKFLOW_YAML = """\ +name: patient-sync +jobs: + fetch-patients: + id: 5f2f36e7-3f42-4b0a-9c11-9b3f5a3d1a01 + name: Fetch Patients + adaptor: "@openfn/language-http@latest" + body: | + get('/patients'); + notify-admin: + id: 8a1c22d0-6e4f-49d3-b6a2-4bfeb1f0c902 + name: Notify Admin + adaptor: "@openfn/language-http@latest" + body: | + fn(state => { + if (!state.data || state.data.length === 0) { + console.warn('SYNC-WARN-042: no patient recrods found'); + } + return state; + }); +triggers: + webhook: + id: 2d7f80b3-1c55-47a9-8e2f-6a90d24c7a03 + type: webhook +edges: + webhook->fetch-patients: + id: c4e9a1f6-0d82-4c37-9b54-7e315f68bd04 + source_trigger: webhook + target_job: fetch-patients + condition_type: always +""" + + +def job_chat_payload(content: str) -> dict: + return { + "content": content, + "suggest_code": True, + "subagent": True, + "workflow_yaml": WORKFLOW_YAML, + "context": { + "expression": "get('/patients');", + "adaptor": "@openfn/language-http@latest", + "page_name": "Fetch Patients", + "job_key": "fetch-patients", + }, + } + + +def workflow_chat_payload(content: str) -> dict: + return { + "content": content, + "subagent": True, + "existing_yaml": WORKFLOW_YAML, + } + + +def test_job_chat_hands_over_structural_request() -> None: + """A workflow-structure request misrouted to job_chat must hand over, + with no user-visible reply text and no code attached.""" + result = job_chat_main(job_chat_payload( + "Add a new step after this one that sends the patients to Salesforce, and connect it up", + )) + + assert result.get("handover") + assert result["response"] == "" + assert result.get("suggested_code") is None + + +def test_job_chat_hands_over_oblique_other_step_edit() -> None: + """An edit whose target code lives in another step, described by what it + does rather than by name, must hand over — not be bodged into the focused + step or met with 'I can't see that code'.""" + result = job_chat_main(job_chat_payload( + "The warning we log when no patients are found should include the run date too, can you update it?", + )) + + assert result.get("handover") + assert result["response"] == "" + assert result.get("suggested_code") is None + + +def test_workflow_chat_hands_over_code_request() -> None: + """A job-code request misrouted to workflow_chat must hand over, + with no user-visible reply text and no YAML.""" + result = workflow_chat_main(workflow_chat_payload( + "Why does the code in my fetch-patients step return an empty array? Can you fix it?", + )) + + assert result.get("handover") + assert result["response"] == "" + assert not result.get("response_yaml") + + +def test_job_chat_hands_over_when_focused_step_is_wrong() -> None: + """Right subagent, wrong step: the user says "this step" but the code they + describe lives in a different step than the one the router focused. The + model must not claim it can't see the warning, and must not bodge a new + warning into the wrong step.""" + result = job_chat_main(job_chat_payload( + "Fix the typo in the warning message this step logs when there are no patients", + )) + + assert result.get("handover") + assert result["response"] == "" + assert result.get("suggested_code") is None + + +def test_workflow_chat_hands_over_oblique_code_change() -> None: + """A code-level change described without naming any step must hand over — + workflow_chat sees only redacted job bodies and cannot make it.""" + result = workflow_chat_main(workflow_chat_payload( + "Can you change the wording of the warning we log when no patients come back?", + )) + + assert result.get("handover") + assert result["response"] == "" + assert not result.get("response_yaml") diff --git a/services/global_chat/tests/unit/test_router.py b/services/global_chat/tests/unit/test_router.py index 1a3ada34..8f97c499 100644 --- a/services/global_chat/tests/unit/test_router.py +++ b/services/global_chat/tests/unit/test_router.py @@ -2,8 +2,8 @@ from unittest.mock import patch -from global_chat.router import RouterAgent -from global_chat.yaml_utils import workflow_has_job_code +from global_chat.router import RouterAgent, RouterDecision, RouterResult +from yaml_utils import workflow_has_job_code EMPTY_YAML = """\ name: wf @@ -33,6 +33,8 @@ def make_router() -> RouterAgent: router._input_attachments = [] router._user = None router._metrics_opt_in = None + router._stream_manager = None + router.model = "claude-test" return router @@ -87,3 +89,100 @@ def test_routing_message_tags_empty_workflow() -> None: router = make_router() msg = router._build_routing_message("what does this do", EMPTY_YAML, None, []) assert "[All step bodies are empty/placeholder]" in msg + + +def test_job_route_sends_subagent_payload() -> None: + router = make_router() + + with patch("job_chat.job_chat.main", return_value=job_chat_result(None)) as mock_main: + router._route_to_job_chat( + "explain this", WORKFLOW_YAML, "workflows/wf/fetch-patients", [], False, 5, + ) + + payload = mock_main.call_args[0][0] + assert payload["subagent"] is True + assert payload["workflow_yaml"] == WORKFLOW_YAML + assert payload["context"]["job_key"] == "fetch-patients" + + +def make_planner_result() -> RouterResult: + return RouterResult( + response="planner answer", + response_segments=[{"type": "text", "content": "planner answer"}], + attachments=[], + history=[], + usage={"input_tokens": 10}, + meta={"agents": ["router", "planner"]}, + ) + + +def test_job_route_handover_reroutes_to_planner() -> None: + router = make_router() + handed_over = {"response": "", "handover": "needs structure changes", "history": [], "usage": {"input_tokens": 7}} + + with patch("job_chat.job_chat.main", return_value=handed_over), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()) as planner_mock: + result = router._route_to_job_chat( + "add a step", WORKFLOW_YAML, "workflows/wf/fetch-patients", [], False, 5, + ) + + planner_mock.assert_called_once() + assert result.response == "planner answer" + # Reroute diagnostics stay out of the response meta (Langfuse-only) + assert "handover_from" not in result.meta + # Usage from the aborted job_chat call is kept on top of the planner's + assert result.usage["input_tokens"] == 17 + + +def test_workflow_route_handover_reroutes_to_planner() -> None: + router = make_router() + handed_over = { + "response": "", "response_yaml": None, "handover": "asks about job code", + "history": [], "usage": {"input_tokens": 3}, + } + + with patch("workflow_chat.workflow_chat.main", return_value=handed_over), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()) as planner_mock: + result = router._route_to_workflow_chat( + "what does this code do", WORKFLOW_YAML, "workflows/wf", [], False, 4, + ) + + planner_mock.assert_called_once() + assert "handover_from" not in result.meta + assert result.usage["input_tokens"] == 13 + + +def test_low_confidence_direct_route_goes_to_planner() -> None: + router = make_router() + decision = RouterDecision(destination="job_code_agent", confidence=2, job_key="fetch-patients") + + with patch.object(RouterAgent, "_make_routing_decision", return_value=decision), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()) as planner_mock, \ + patch.object(RouterAgent, "_route_to_job_chat") as job_mock: + result = router.route_and_execute("edit this", WORKFLOW_YAML, None, [], False) + + planner_mock.assert_called_once() + job_mock.assert_not_called() + assert result.response == "planner answer" + + +def test_confident_direct_route_is_not_gated() -> None: + router = make_router() + decision = RouterDecision(destination="job_code_agent", confidence=3, job_key="fetch-patients") + job_result = RouterResult( + response="job answer", + response_segments=[{"type": "text", "content": "job answer"}], + attachments=[], + history=[], + usage={}, + meta={}, + ) + + with patch.object(RouterAgent, "_make_routing_decision", return_value=decision), \ + patch.object(RouterAgent, "_route_to_planner") as planner_mock, \ + patch.object(RouterAgent, "_route_to_job_chat", return_value=job_result) as job_mock: + result = router.route_and_execute("edit this", WORKFLOW_YAML, None, [], False) + + job_mock.assert_called_once() + planner_mock.assert_not_called() + assert result.response == "job answer" diff --git a/services/global_chat/tests/unit/test_yaml_utils.py b/services/global_chat/tests/unit/test_yaml_utils.py new file mode 100644 index 00000000..024965f4 --- /dev/null +++ b/services/global_chat/tests/unit/test_yaml_utils.py @@ -0,0 +1,68 @@ +"""Unit tests for the shared inspect_job_code tool executor and redaction.""" + +from yaml_utils import inspect_job_code, redact_job_bodies + +WORKFLOW_YAML = """\ +name: wf +jobs: + fetch-patients: + name: Fetch Patients + body: get('/patients'); + send-data: + name: Send Data + body: post('/data', $.data); +""" + +WORKFLOW_YAML_WITH_IDS = """\ +name: wf +jobs: + fetch-patients: + id: 5f2f36e7-3f42-4b0a-9c11-9b3f5a3d1a01 + name: Fetch Patients + adaptor: "@openfn/language-http@latest" + body: get('/patients'); +triggers: + webhook: + id: 2d7f80b3-1c55-47a9-8e2f-6a90d24c7a03 + type: webhook +edges: + webhook->fetch-patients: + id: c4e9a1f6-0d82-4c37-9b54-7e315f68bd04 + source_trigger: webhook + target_job: fetch-patients + condition_type: always +""" + + +def test_redact_strips_bodies_and_ids_keeps_structure() -> None: + redacted = redact_job_bodies(WORKFLOW_YAML_WITH_IDS) + + assert "get('/patients');" not in redacted + assert "# [use inspect_job_code to view]" in redacted + assert "id:" not in redacted + # Structure the model needs stays intact + assert "Fetch Patients" in redacted + assert "@openfn/language-http@latest" in redacted + assert "webhook->fetch-patients" in redacted + assert "condition_type: always" in redacted + + +def test_inspect_returns_requested_bodies() -> None: + result = inspect_job_code(WORKFLOW_YAML, ["fetch-patients", "send-data"]) + assert "get('/patients');" in result + assert "post('/data', $.data);" in result + + +def test_inspect_matches_fuzzy_names() -> None: + result = inspect_job_code(WORKFLOW_YAML, ["Fetch Patients"]) + assert "get('/patients');" in result + + +def test_inspect_reports_missing_job() -> None: + result = inspect_job_code(WORKFLOW_YAML, ["nonexistent"]) + assert "No code found for job 'nonexistent'." in result + + +def test_inspect_handles_missing_yaml_and_keys() -> None: + assert inspect_job_code(None, ["a"]) == "No workflow available to inspect." + assert inspect_job_code(WORKFLOW_YAML, []) == "ERROR: No job keys provided." diff --git a/services/global_chat/tools/tool_definitions.py b/services/global_chat/tools/tool_definitions.py index 6f94fa7b..6322e20e 100644 --- a/services/global_chat/tools/tool_definitions.py +++ b/services/global_chat/tools/tool_definitions.py @@ -73,24 +73,9 @@ "cache_control": {"type": "ephemeral"} } -# Tool 4: Inspect job code -INSPECT_JOB_CODE_TOOL = { - "name": "inspect_job_code", - "description": """Read the current code body of one or more jobs in the workflow (read-only). - -Use this to inspect existing step code before editing — e.g. to find which steps a change applies to before editing only those, or to base one step on another. Pass all the job keys you need in a single call rather than calling once per job.""", - "input_schema": { - "type": "object", - "properties": { - "job_keys": { - "type": "array", - "items": {"type": "string"}, - "description": "The job keys to inspect (e.g. ['fetch-patients', 'load-dhis2'])" - } - }, - "required": ["job_keys"] - } -} +# Tool 4: Inspect job code — shared with job_chat's subagent mode so both +# agents explore the workflow with the exact same tool +from yaml_utils import INSPECT_JOB_CODE_TOOL # noqa: E402 # Export all tool definitions TOOL_DEFINITIONS = [ diff --git a/services/global_chat/yaml_utils.py b/services/global_chat/yaml_utils.py deleted file mode 100644 index 119bbcf0..00000000 --- a/services/global_chat/yaml_utils.py +++ /dev/null @@ -1,123 +0,0 @@ -""" -Shared YAML utility functions for working with workflow YAML strings. - -Used by router and subagent caller for job extraction and code stitching. -""" -import re -import yaml -from typing import Dict, Optional, Tuple - - -def get_step_name_from_page(page: Optional[str]) -> Optional[str]: - """ - Extract step name from page URL. - - Examples: - workflows/my-workflow/fetch-patients -> "fetch-patients" - workflows/my-workflow -> None - workflows/my-workflow/settings -> None - """ - if not page: - return None - - parts = page.strip("/").split("/") - if len(parts) == 3 and parts[0] == "workflows" and parts[2] != "settings": - return parts[2] - - return None - - -def normalize_name(name: str) -> str: - """Normalize a name for fuzzy matching: lowercase, non-alphanumeric chars become hyphens.""" - return re.sub(r'[^a-z0-9]', '-', name.lower()).strip('-') - - -def find_job_in_yaml(yaml_str: str, step_name: str) -> Tuple[Optional[str], Optional[Dict]]: - """ - Find a job in the workflow YAML by step name. - - Tries direct key match first, then normalized name comparison against - both the job key and the job's name field. - - Returns: - (job_key, job_data) or (None, None) if not found or on parse error - """ - try: - yaml_data = yaml.safe_load(yaml_str) - except Exception: - return None, None - - if not yaml_data or "jobs" not in yaml_data: - return None, None - - jobs = yaml_data["jobs"] - - # Direct key match - if step_name in jobs: - return step_name, jobs[step_name] - - # Normalized match: compare against job key and name field - normalized_step = normalize_name(step_name) - for job_key, job_data in jobs.items(): - if normalize_name(job_key) == normalized_step: - return job_key, job_data - job_name = job_data.get("name", "") - if normalize_name(job_name) == normalized_step: - return job_key, job_data - - return None, None - - -EMPTY_JOB_BODY = "// Add operations here" - - -def workflow_has_job_code(yaml_str: Optional[str]) -> bool: - """Return True if any job has a non-empty, non-placeholder body. - - The canonical empty-job marker is ``// Add operations here`` (see - workflow_chat); a blank body or that marker means "no code yet". Used to - decide whether a "what does this do" question needs the planner (to read the - real code) or can take the faster workflow_agent path (structure only). - """ - try: - yaml_data = yaml.safe_load(yaml_str) - except Exception: - return False - if not yaml_data or "jobs" not in yaml_data: - return False - for job_data in yaml_data["jobs"].values(): - body = (job_data or {}).get("body") - if isinstance(body, str) and body.strip() and body.strip() != EMPTY_JOB_BODY: - return True - return False - - -def redact_job_bodies(yaml_str: str) -> str: - """Return workflow YAML with job bodies replaced by a placeholder.""" - try: - yaml_data = yaml.safe_load(yaml_str) - if yaml_data and "jobs" in yaml_data: - for job_data in yaml_data["jobs"].values(): - if "body" in job_data: - job_data["body"] = "# [use inspect_job_code to view]" - return yaml.dump(yaml_data, sort_keys=False) - except Exception: - pass - return yaml_str - - -def stitch_job_code(yaml_str: str, job_key: str, new_code: str) -> str: - """ - Replace a job's body in the workflow YAML with new code. - - Returns the original YAML string unchanged if parsing or stitching fails. - """ - try: - yaml_data = yaml.safe_load(yaml_str) - if yaml_data and "jobs" in yaml_data and job_key in yaml_data["jobs"]: - yaml_data["jobs"][job_key]["body"] = new_code - return yaml.dump(yaml_data, sort_keys=False) - except Exception: - pass - - return yaml_str diff --git a/services/job_chat/job_chat.py b/services/job_chat/job_chat.py index 8a52f4fd..368f8573 100644 --- a/services/job_chat/job_chat.py +++ b/services/job_chat/job_chat.py @@ -20,6 +20,7 @@ from langfuse import observe, propagate_attributes, get_client as get_langfuse_client from langfuse_util import should_track, build_tags, build_generation_diff from util import ApolloError, create_logger, AdaptorSpecifier, add_page_prefix, APOLLO_VERSION +from yaml_utils import INSPECT_JOB_CODE_TOOL, inspect_job_code from .prompt import build_prompt, build_error_correction_prompt from .old_prompt import build_old_prompt from streaming_util import ( @@ -82,6 +83,49 @@ }, } +# Subagent mode only (job_chat called from global_chat): escalation disguised +# as a capability. Calling it hands the request back to the caller, which +# reroutes to the planner — so if the model narrates before calling, the +# narration ("I'll take a look at your workflow") matches what happens next. +_EDIT_WORKFLOW_TOOL = { + "name": "edit_workflow", + "description": ( + "Open the full workflow to work on anything beyond this step's code: " + "workflow structure (add/remove/rename steps, triggers, edges, adaptors) " + "or code changes in other steps. Call this as your VERY FIRST action. " + "To merely READ another step's code, use inspect_job_code instead." + ), + "input_schema": { + "type": "object", + "properties": { + "goal": { + "type": "string", + "description": "One sentence: what needs to be done", + } + }, + "required": ["goal"], + "additionalProperties": False, + }, +} + +# The planner's inspect tool (same name, schema, and executor via yaml_utils), +# with the description rewritten for job_chat: unlike the planner, job_chat can +# only edit the focused step, so reading must never look like a way to act. +_INSPECT_JOB_CODE_TOOL = { + **INSPECT_JOB_CODE_TOOL, + "description": ( + "Read the current code of one or more other steps in the workflow. " + "Use it when seeing another step's code helps you answer a question or " + "edit the focused step — e.g. to match its pattern, or to see the state " + "shape it produces. To change another step's code, call edit_workflow " + "instead. Pass all the job keys you need in a single call." + ), +} + +# Max API rounds in one generate() call: enough for a couple of +# inspect_job_code round-trips plus the final answer. +_MAX_TOOL_ROUNDS = 4 + # Helper function for page navigation def extract_page_prefix_from_last_turn(history: List[Dict[str, str]]) -> Optional[str]: @@ -116,6 +160,11 @@ class Payload: download_adaptor_docs: Optional[bool] = True refresh_rag: Optional[bool] = False metrics_opt_in: Optional[bool] = None + # Subagent mode: set only when called from global_chat, never by direct + # production callers. workflow_yaml additionally enables the + # inspect_job_code tool. + workflow_yaml: Optional[str] = None + subagent: Optional[bool] = False @classmethod def from_dict(cls, data: Dict[str, Any]) -> "Payload": @@ -135,6 +184,8 @@ def from_dict(cls, data: Dict[str, Any]) -> "Payload": download_adaptor_docs=data.get("download_adaptor_docs", True), refresh_rag=data.get("refresh_rag", False), metrics_opt_in=data.get("metrics_opt_in"), + workflow_yaml=data.get("workflow_yaml"), + subagent=data.get("subagent", False), ) @@ -153,6 +204,8 @@ class ChatResponse: usage: Dict[str, Any] rag: Dict[str, Any] diff: Optional[Dict[str, Any]] = None + # Subagent mode only: reason the request was handed back to the caller + handover: Optional[str] = None class AnthropicClient: def __init__(self, config: Optional[ChatConfig] = None): @@ -186,17 +239,25 @@ def generate( stream: Optional[bool] = False, download_adaptor_docs: Optional[bool] = True, refresh_rag: Optional[bool] = False, - current_page: Optional[dict] = None + current_page: Optional[dict] = None, + workflow_yaml: Optional[str] = None, + subagent: Optional[bool] = False, + stream_manager: Optional[StreamManager] = None, ) -> ChatResponse: """ Generate a response using the Claude API with optional streaming. + + In subagent mode (called from global_chat) the model can also read + other steps' code via inspect_job_code and hand the request back via + handover. A stream_manager may be injected by the caller so a handed- + over request continues on the same stream. """ sentry_sdk.set_tag("prompt_type", "code_suggestions" if suggest_code else "no_code_suggestions") with sentry_sdk.start_transaction(name="chat_generation") as transaction: history = history.copy() if history else [] - stream_manager = StreamManager(model=self.config.model, stream=stream) + stream_manager = stream_manager or StreamManager(model=self.config.model, stream=stream) if context and context.get("expression"): stream_manager.send_thinking(STATUS_REVIEWING_CODE) else: @@ -212,7 +273,9 @@ def generate( api_key=self.api_key, stream_manager=stream_manager, download_adaptor_docs=download_adaptor_docs, - refresh_rag=refresh_rag + refresh_rag=refresh_rag, + workflow_yaml=workflow_yaml, + subagent=subagent ) else: @@ -230,74 +293,148 @@ def generate( # tool. tool_choice stays "auto": the model answers in text # and only calls the tool when it actually wants to change the job. output_config = {"effort": "medium"} - tool_kwargs = ( - {"tools": [_EDIT_TOOL], "tool_choice": {"type": "auto"}} - if suggest_code else {} - ) + tools = [] + if suggest_code: + tools.append(_EDIT_TOOL) + if subagent: + tools.append(_EDIT_WORKFLOW_TOOL) + if workflow_yaml: + tools.append(_INSPECT_JOB_CODE_TOOL) + tool_kwargs = {"tools": tools, "tool_choice": {"type": "auto"}} if tools else {} + + # Without the subagent tools this loop runs exactly once: edit_job + # is terminal (its input IS the output), so only inspect_job_code + # triggers another round and only handover exits early. + messages = prompt + handover_reason = None + text_parts = [] + usage_events = [] with sentry_sdk.start_span(description="anthropic_api_call"): - if stream: - logger.info("Making streaming API call") - text_started = False - sent_length = 0 - accumulated_response = "" - self._stream_applied = False - self._stream_suggested_code = None - self._stream_diff = None - - original_code = context.get("expression") if context and isinstance(context, dict) else None - - stream_kwargs = dict( - max_tokens=self.config.max_tokens, - messages=prompt, - model=self.config.model, - system=system_message, - thinking={"type": "adaptive"}, - output_config=output_config, - **tool_kwargs - ) + for round_index in range(_MAX_TOOL_ROUNDS): + if stream: + logger.info("Making streaming API call") + text_started = False + sent_length = 0 + accumulated_response = "" + self._stream_applied = False + self._stream_suggested_code = None + self._stream_diff = None + + original_code = context.get("expression") if context and isinstance(context, dict) else None + + stream_kwargs = dict( + max_tokens=self.config.max_tokens, + messages=messages, + model=self.config.model, + system=system_message, + thinking={"type": "adaptive"}, + output_config=output_config, + **tool_kwargs + ) - with self.client.messages.stream(**stream_kwargs) as stream_obj: - for event in stream_obj: - if event.type == "message_start": - stream_manager.send_thinking(STATUS_WORKING) - # The edit_job tool block starts after the text ends; its - # input (the code) streams silently, so show a status here. - elif event.type == "content_block_start" and getattr(getattr(event, "content_block", None), "type", None) == "tool_use": - stream_manager.send_thinking(STATUS_WRITING_CODE) - accumulated_response, text_started, sent_length = self.process_stream_event( - event, - accumulated_response, - suggest_code, - text_started, - sent_length, - stream_manager, - original_code, - content - ) - message = stream_obj.get_final_message() + with self.client.messages.stream(**stream_kwargs) as stream_obj: + for event in stream_obj: + if event.type == "message_start" and round_index == 0: + stream_manager.send_thinking(STATUS_WORKING) + # The edit_job tool block starts after the text ends; its + # input (the code) streams silently, so show a status here. + elif event.type == "content_block_start" and getattr(getattr(event, "content_block", None), "type", None) == "tool_use" and getattr(getattr(event, "content_block", None), "name", None) == "edit_job": + stream_manager.send_thinking(STATUS_WRITING_CODE) + accumulated_response, text_started, sent_length = self.process_stream_event( + event, + accumulated_response, + suggest_code, + text_started, + sent_length, + stream_manager, + original_code, + content + ) + message = stream_obj.get_final_message() + + # Flush any remaining buffered text, stripping JSON closing chars + if suggest_code and text_started: + if sent_length < len(accumulated_response): + remaining = accumulated_response[sent_length:] + remaining = re.sub(r'"\s*}\s*$', '', remaining) + if remaining: + stream_manager.send_text(self._unescape_json_string(remaining)) + + else: + logger.info("Making non-streaming API call") + create_kwargs = dict( + max_tokens=self.config.max_tokens, messages=messages, model=self.config.model, system=system_message, + thinking={"type": "adaptive"}, + output_config=output_config, + # Per-request timeout (same values as the SDK default): + # required for non-streaming calls with max_tokens > ~21k, + # which the SDK otherwise rejects. + timeout=httpx.Timeout(600.0, connect=5.0), + **tool_kwargs + ) + message = self.client.messages.create(**create_kwargs) - # Flush any remaining buffered text, stripping JSON closing chars - if suggest_code and text_started: - if sent_length < len(accumulated_response): - remaining = accumulated_response[sent_length:] - remaining = re.sub(r'"\s*}\s*$', '', remaining) - if remaining: - stream_manager.send_text(self._unescape_json_string(remaining)) + if hasattr(message, "usage"): + usage_events.append(message.usage.model_dump()) - else: - logger.info("Making non-streaming API call") - create_kwargs = dict( - max_tokens=self.config.max_tokens, messages=prompt, model=self.config.model, system=system_message, - thinking={"type": "adaptive"}, - output_config=output_config, - # Per-request timeout (same values as the SDK default): - # required for non-streaming calls with max_tokens > ~21k, - # which the SDK otherwise rejects. - timeout=httpx.Timeout(600.0, connect=5.0), - **tool_kwargs - ) - message = self.client.messages.create(**create_kwargs) + for content_block in message.content: + if getattr(content_block, "type", None) == "text" and content_block.text: + text_parts.append(content_block.text) + + tool_uses = [b for b in message.content if getattr(b, "type", None) == "tool_use"] + if tool_uses: + logger.info( + "job_chat round %d: model called %s", + round_index, ", ".join(b.name for b in tool_uses), + ) + + handover_block = next((b for b in tool_uses if b.name == "edit_workflow"), None) + if handover_block: + handover_reason = (handover_block.input or {}).get("goal") or "handover requested" + break + + inspect_blocks = [b for b in tool_uses if b.name == "inspect_job_code"] + if not inspect_blocks or round_index == _MAX_TOOL_ROUNDS - 1: + break + + # Answer the inspect calls and let the model continue. An + # edit_job call made in the same round is deferred: the + # model must re-issue it with its final answer. + stream_manager.send_thinking(STATUS_REVIEWING_CODE) + tool_results = [] + for block in tool_uses: + if block.name == "inspect_job_code": + job_keys = (block.input or {}).get("job_keys") or [] + logger.info("job_chat inspect_job_code: reading %s", job_keys) + result_text = inspect_job_code(workflow_yaml, job_keys) + else: + result_text = ( + "Not applied. Finish inspecting, then write your final reply " + "and call edit_job again with the complete edits." + ) + tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": result_text}) + + messages = messages + [ + {"role": "assistant", "content": message.content}, + {"role": "user", "content": tool_results}, + ] + + if handover_reason: + logger.info(f"job_chat handing over: {handover_reason}") + # Deliberately do NOT end the stream: the caller reroutes the + # request and the next agent continues on the same stream. + return ChatResponse( + response="", + suggested_code=None, + history=history, + usage=self.sum_usage( + *usage_events, + *[usage_data for usage_key, usage_data in retrieved_knowledge.get("usage", {}).items()] + ), + rag=retrieved_knowledge, + handover=handover_reason, + ) if hasattr(message, "usage"): if message.usage.cache_creation_input_tokens: @@ -306,15 +443,13 @@ def generate( logger.info(f"Cache read: {message.usage.cache_read_input_tokens} tokens") # The model answers in normal text; it calls the `edit_job` tool only - # when it wants to change the user's job. So text = the reply, and the - # tool's parsed input carries the code edits (no JSON-in-text parsing). - text_parts = [] + # when it wants to change the user's job. So text = the reply + # (accumulated across tool rounds above), and the tool's parsed + # input carries the code edits (no JSON-in-text parsing). tool_code_edits = None for content_block in message.content: if getattr(content_block, "type", None) == "tool_use" and getattr(content_block, "name", None) == "edit_job": tool_code_edits = (content_block.input or {}).get("code_edits") or [] - elif getattr(content_block, "type", None) == "text": - text_parts.append(content_block.text) text_response = "\n\n".join(text_parts).strip() suggested_code = None @@ -355,7 +490,7 @@ def generate( ] usage = self.sum_usage( - message.usage.model_dump() if hasattr(message, "usage") else {}, + *usage_events, *[usage_data for usage_key, usage_data in retrieved_knowledge.get("usage", {}).items()] ) @@ -608,7 +743,7 @@ def main(data_dict: dict) -> dict: """ try: sentry_sdk.set_context("request_data", { - k: v for k, v in data_dict.items() if k != "api_key" + k: v for k, v in data_dict.items() if k not in ("api_key", "_stream_manager") }) data = Payload.from_dict(data_dict) @@ -667,7 +802,12 @@ def main(data_dict: dict) -> dict: stream=data.stream, download_adaptor_docs=data.download_adaptor_docs, refresh_rag=should_refresh_rag, - current_page=current_page + current_page=current_page, + workflow_yaml=data.workflow_yaml, + subagent=data.subagent, + # In-process callers (global_chat) may inject a shared stream + # manager so a handed-over request continues the same stream + stream_manager=data_dict.get("_stream_manager"), ) # Tag the trace when code was generated, so we can filter for it. @@ -675,6 +815,12 @@ def main(data_dict: dict) -> dict: with propagate_attributes(tags=["has_code_attachment"]): pass + # Tag the trace when the request was handed back for rerouting to + # the planner, so we can filter for handovers. + if tracking and result.handover: + with propagate_attributes(tags=["handover"]): + pass + if tracking: diff_meta = build_generation_diff( original=data.context.get("expression"), @@ -694,6 +840,9 @@ def main(data_dict: dict) -> dict: if result.diff: response_dict["diff"] = result.diff + if result.handover: + response_dict["handover"] = result.handover + return response_dict except ApolloError: diff --git a/services/job_chat/prompt.py b/services/job_chat/prompt.py index 1dd27dfc..fb1afc11 100644 --- a/services/job_chat/prompt.py +++ b/services/job_chat/prompt.py @@ -4,12 +4,13 @@ import sentry_sdk from langfuse import observe from util import create_logger, ApolloError, AdaptorSpecifier, get_db_connection +from yaml_utils import redact_job_bodies, normalize_name from .retrieve_docs import retrieve_knowledge from search_adaptor_docs.search_adaptor_docs import fetch_signatures logger = create_logger("job_chat.prompt") -system_role = """ +_role_before_scope = """ You are a software engineer helping a non-expert user write a job for our platform. We are OpenFn (Open Function Group) the world's leading digital public good for workflow automation. @@ -32,10 +33,17 @@ Your chat panel is embedded in a web based IDE, which lets users build a Workflow with a number of steps (or jobs). There is a code editor next to you, which users can copy and paste code into. Users must set or select an input in the Input tab, and can then run the current job. +""" +# Production only. In global chat's subagent mode, structure requests are +# handled via edit_workflow, so this scope restriction is omitted entirely +# rather than contradicted. +production_scope_instructions = """ You ONLY help with job code. Do NOT help with overall workflow structure. If the user wants to add/remove/edit workflow steps, tell them to navigate to the workflow overview. +""" +_role_after_scope = """ Users can Flag any answers that are not helpful, which will help us build a better prompt for you. @@ -60,6 +68,9 @@ """ +system_role = _role_before_scope + production_scope_instructions + _role_after_scope +subagent_system_role = _role_before_scope + _role_after_scope + job_writing_summary = """ When writing jobs, users will use their own credentials to access different @@ -165,6 +176,26 @@ """ +# Appended in subagent mode only (when job_chat is called from global_chat). +subagent_mode_instructions = """ + +The workflow has other steps, but your edit_job tool edits THIS step only. +Decide by where the change lands: + +- Change lands in THIS step (or nothing needs changing): handle it here. Read + any other step with `inspect_job_code` whenever it helps — what an upstream + step outputs, keeping style or field names consistent, finding code the user + mentions that is not in . +- Change lands anywhere else — workflow structure (add/remove/rename steps, + triggers, edges, adaptors) or another step's code, even code the user calls + "this step": call `edit_workflow` as your very first action, before any + reply text (at most exactly: "I'll take a look at your workflow."). + +If the user mentions code you can't find in , assume it lives in +another step — never reply that it isn't here. + +""" + # Response contract, appended last in the system message. output_format = """ @@ -261,10 +292,37 @@ def has(self, key): return hasattr(self, key) and getattr(self, key) is not None -def generate_system_message(context_dict, search_results, download_adaptor_docs=True, stream_manager=None): +def build_focus_line(viewing, focused): + """One sentence orienting the model to what the user has on screen, and which + step it can edit when that differs. + + `viewing` (router-only) is the on-screen step name, the literal "canvas", or + None when the caller can't tell; `focused` is the editable step. When both + are a step and coincide (the common case) they fuse into one clause; they + diverge only when the request targets a step other than the one open. + + Returns "" when there is nothing worth stating (no viewing info): the model + works from and the tools, and we avoid narrating editing plumbing + the model could echo back to the user. + """ + if viewing and viewing != "canvas": + if not focused: + return f"The user has the '{viewing}' step's code open." + if normalize_name(viewing) == normalize_name(focused): + return f"The user has the '{viewing}' step's code open — that's the step you're currently editing." + return f"The user has the '{viewing}' step open, but the step you're editing is '{focused}' — likely what their request is about." + if viewing == "canvas": + if focused: + return f"The user is viewing the workflow canvas, not a specific step; the '{focused}' step is the one you're currently editing." + return "The user is viewing the workflow canvas." + return "" + + +def generate_system_message(context_dict, search_results, download_adaptor_docs=True, stream_manager=None, + workflow_yaml=None, subagent=False): context = context_dict if isinstance(context_dict, Context) else Context(**(context_dict or {})) - message = [system_role] + message = [subagent_system_role if subagent else system_role] message.append(f"{job_writing_summary}") message.append({"type": "text", "text": ".", "cache_control": {"type": "ephemeral"}}) @@ -353,6 +411,27 @@ def generate_system_message(context_dict, search_results, download_adaptor_docs= ```{context.log}``` """) + if subagent: + message.append(subagent_mode_instructions) + if workflow_yaml: + # `focused` is the editable step; `viewing` (router-only) is what the + # user actually has on screen — a step's code or the literal "canvas". + # Grounding focus in their view lets a bare "this step" resolve to + # what they're looking at. Both may be absent (planner/prod, or an + # unresolved page), in which case build_focus_line returns "". + focused = context.job_key if context.has("job_key") else ( + context.page_name if context.has("page_name") else None) + viewing = context.viewing if context.has("viewing") else None + focus = build_focus_line(viewing, focused) + header = ["The full workflow, job code redacted."] + if focus: + header.append(focus) + header.append("READ other steps' code with `inspect_job_code` when the request refers to them.") + redacted = redact_job_bodies(workflow_yaml) + message.append( + f"\n{' '.join(header)}\n\n{redacted}\n" + ) + # Output contract goes LAST so it is the final, most prominent instruction. message.append(output_format) @@ -365,7 +444,8 @@ def format_search_results(search_results): ]) @observe(name="job_chat_build_prompt") -def build_prompt(content, history, context, rag=None, api_key=None, stream_manager=None, download_adaptor_docs=True, refresh_rag=False): +def build_prompt(content, history, context, rag=None, api_key=None, stream_manager=None, download_adaptor_docs=True, refresh_rag=False, + workflow_yaml=None, subagent=False): retrieved_knowledge = { "search_results": [], "search_results_sections": [], @@ -398,7 +478,9 @@ def build_prompt(content, history, context, rag=None, api_key=None, stream_manag context_dict=context, search_results=retrieved_knowledge.get("search_results") if retrieved_knowledge is not None else None, download_adaptor_docs=download_adaptor_docs, - stream_manager=stream_manager) + stream_manager=stream_manager, + workflow_yaml=workflow_yaml, + subagent=subagent) prompt = [] prompt.extend(history) @@ -407,9 +489,15 @@ def build_prompt(content, history, context, rag=None, api_key=None, stream_manag # only remind the model to route an actual code change through `edit_job`. # Added only to the message sent to the model; the stored history (built in # generate() from the raw content) omits it, so it never accumulates. + reminder = "Reply in text. If this requires changing the job code, also call the `edit_job` tool to apply the change." + if subagent: + reminder += ( + " If it needs changes beyond this step's code, call `edit_workflow` first;" + " to merely read another step (to answer, or to edit this one), use `inspect_job_code`." + ) prompt.append({ "role": "user", - "content": f"{content}\n\nReply in text. If this requires changing the job code, also call the `edit_job` tool to apply the change.", + "content": f"{content}\n\n{reminder}", }) return (system_message, prompt, retrieved_knowledge) diff --git a/services/job_chat/tests/unit/test_subagent_prompt.py b/services/job_chat/tests/unit/test_subagent_prompt.py new file mode 100644 index 00000000..a75fc808 --- /dev/null +++ b/services/job_chat/tests/unit/test_subagent_prompt.py @@ -0,0 +1,77 @@ +"""Unit tests for job_chat's subagent-mode system prompt.""" + +from job_chat.prompt import generate_system_message + +WORKFLOW_YAML = """\ +name: wf +jobs: + fetch-patients: + name: Fetch Patients + body: get('/patients'); +""" + + +def system_text(**kwargs) -> str: + blocks = generate_system_message(context_dict={}, search_results=None, **kwargs) + return "\n".join(b["text"] for b in blocks) + + +def test_production_prompt_keeps_navigate_instruction(): + text = system_text() + + # Production callers never set subagent: the only-job-code scope and the + # go-to-the-workflow-overview instruction must stay untouched, and no + # subagent sections appear + assert "tell them to navigate to the workflow overview" in text + assert "Do NOT help with overall workflow structure" in text + assert "edit_workflow" not in text + assert "" not in text + + +def test_subagent_prompt_strips_navigate_instruction(): + text = system_text(subagent=True, workflow_yaml=WORKFLOW_YAML) + + # Neither the go-elsewhere phrasing nor the only-job-code scope may be in + # context at all (production_scope_instructions is omitted from the + # composed subagent_system_role) + assert "navigate to the workflow overview" not in text + assert "Do NOT help with overall workflow structure" not in text + assert "ONLY help with job code" not in text + assert "edit_workflow" in text + assert "" in text + + +def _subagent_text(context: dict) -> str: + blocks = generate_system_message( + context_dict=context, search_results=None, + subagent=True, workflow_yaml=WORKFLOW_YAML, + ) + return "\n".join(b["text"] for b in blocks) + + +def test_subagent_prompt_grounds_focus_in_viewed_step(): + # No viewing info: no focus sentence is emitted (the model works from + # ), and no editing-plumbing phrasing leaks. Structure still shown. + text = system_text(subagent=True, workflow_yaml=WORKFLOW_YAML) + assert "" in text + assert "No step is focused" not in text + assert "loaded for editing" not in text + assert "currently editing" not in text + + # Viewing the same step it can edit (common case): named and framed as open. + text = _subagent_text({"job_key": "fetch-patients", "viewing": "Fetch Patients"}) + assert "Fetch Patients" in text + assert "currently editing" in text + + # Viewing the workflow canvas: says "canvas" and still names the editable + # step, with no editing-plumbing phrase. + text = _subagent_text({"job_key": "fetch-patients", "viewing": "canvas"}) + assert "workflow canvas" in text + assert "'fetch-patients'" in text + assert "loaded for editing" not in text + + # Mismatch — viewing one step, editing another: states both, softly. + text = _subagent_text({"job_key": "fetch-patients", "viewing": "Notify Admin"}) + assert "Notify Admin" in text + assert "'fetch-patients'" in text + assert "likely what their request is about" in text diff --git a/services/workflow_chat/gen_project_prompt.py b/services/workflow_chat/gen_project_prompt.py index 36283edb..35eeaf29 100644 --- a/services/workflow_chat/gen_project_prompt.py +++ b/services/workflow_chat/gen_project_prompt.py @@ -30,17 +30,18 @@ def build_system_message(mode_config, existing_yaml=None): return system_message -def build_prompt(content, existing_yaml=None, errors=None, history=None, read_only=False): +def build_prompt(content, existing_yaml=None, errors=None, history=None, read_only=False, subagent=False): """ Build a prompt for the LLM based on mode and context. - + Args: content: User message content existing_yaml: Current YAML being edited (optional) errors: Error messages if in error mode (optional) history: Conversation history (optional) read_only: Whether in read-only mode - + subagent: Whether called from global_chat (adds handover instructions) + Returns: Tuple of (system_message, prompt_messages) """ @@ -75,8 +76,22 @@ def build_prompt(content, existing_yaml=None, errors=None, history=None, read_on user_content = content system_message = build_system_message(mode_config, existing_yaml) - + + if subagent: + # Job-code requests are handed over instead — remove the decline-and- + # navigate-to-the-Inspector instruction (it appears in two prompt + # sections) so it can never slip out. Must match prompts yaml verbatim; + # a unit test guards against the two drifting apart. + system_message = system_message.replace( + "If the user asks for job code, DECLINE to provide it yet, and explain that they " + "need to save their workflow and then navigate to the specific job's code page in " + "the Inspector. Once there, you can help them write the code (and will be able to " + "see any existing code for that job).", + 'If the user asks for job code, set "handover" (see Job Code Requests below).', + ) + system_message += "\n" + config_loader.get_prompt("subagent_handover_instructions") + prompt = list(history) # Create a copy prompt.append({"role": "user", "content": user_content}) - + return (system_message, prompt) \ No newline at end of file diff --git a/services/workflow_chat/gen_project_prompts.yaml b/services/workflow_chat/gen_project_prompts.yaml index 07da00e7..ae5b4ca7 100644 --- a/services/workflow_chat/gen_project_prompts.yaml +++ b/services/workflow_chat/gen_project_prompts.yaml @@ -276,3 +276,15 @@ prompts: Always set the "yaml" key to null. The user's latest message and prior conversation are provided below. Generate your response accordingly. + + subagent_handover_instructions: | + + ## Job Code Requests + + Your response JSON has an extra FIRST field: "handover". + - If the request is chiefly about the code inside a step (reading, explaining, + debugging, or editing it), set "handover" to a short reason, "yaml" to null and + "text" to "" — the request is then rerouted and handled with full code access. + Never decline such requests or tell the user to navigate elsewhere or save + first; this overrides earlier instructions. + - Otherwise set "handover" to null and answer normally. diff --git a/services/workflow_chat/tests/unit/client/test_handover.py b/services/workflow_chat/tests/unit/client/test_handover.py new file mode 100644 index 00000000..732c4d0c --- /dev/null +++ b/services/workflow_chat/tests/unit/client/test_handover.py @@ -0,0 +1,45 @@ +"""Unit tests for subagent-mode handover parsing in workflow_chat.""" + +import json + +from workflow_chat.workflow_chat import AnthropicClient + + +def make_client() -> AnthropicClient: + """Build an AnthropicClient without an API key.""" + client = AnthropicClient.__new__(AnthropicClient) + client._streamed_yaml = None + client._handover = None + return client + + +def test_split_captures_handover_and_skips_yaml() -> None: + client = make_client() + response = json.dumps({"handover": "asks about job code", "yaml": "name: wf", "text": ""}) + + text, output_yaml = client.split_format_yaml(response) + + assert client._handover == "asks about job code" + assert output_yaml == "" + assert text == "" + + +def test_split_without_handover_behaves_normally() -> None: + client = make_client() + response = json.dumps({"handover": None, "yaml": None, "text": "The trigger runs daily."}) + + text, output_yaml = client.split_format_yaml(response) + + assert client._handover is None + assert text == "The trigger runs daily." + assert output_yaml == "" + + +def test_split_legacy_schema_without_handover_field() -> None: + client = make_client() + response = json.dumps({"yaml": None, "text": "Answer."}) + + text, _ = client.split_format_yaml(response) + + assert client._handover is None + assert text == "Answer." diff --git a/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py b/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py index a848a180..b1500d7f 100644 --- a/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py +++ b/services/workflow_chat/tests/unit/gen_project/test_prompt_build.py @@ -33,6 +33,36 @@ def test_build_prompt_error_mode(): assert prompt[-1]["content"] == "Fix the workflow\nThis is the error message:\nInvalid trigger type" +def test_build_prompt_production_keeps_inspector_instruction(): + system_msg, _ = build_prompt( + content="Create a workflow", + existing_yaml="name: test-workflow", + history=[], + ) + + # Production callers never set subagent: the decline-and-navigate + # instruction must stay untouched + assert "navigate to the specific job's code page in the Inspector" in system_msg + assert "handover" not in system_msg + + +def test_build_prompt_subagent_strips_inspector_instruction(): + system_msg, _ = build_prompt( + content="Create a workflow", + existing_yaml="name: test-workflow", + history=[], + subagent=True, + ) + + # The go-elsewhere phrasing must not be in context at all, in any of the + # prompt sections it appears in — if this fails, the sentence in + # gen_project_prompts.yaml and the replace() in build_prompt have drifted + assert "navigate to the specific job's code page in the Inspector" not in system_msg + assert "DECLINE" not in system_msg + assert 'If the user asks for job code, set "handover"' in system_msg + assert "Job Code Requests" in system_msg + + def test_build_prompt_readonly_mode(): system_msg, prompt = build_prompt( content="What does this workflow do?", diff --git a/services/workflow_chat/workflow_chat.py b/services/workflow_chat/workflow_chat.py index a4cf9366..67ee6711 100644 --- a/services/workflow_chat/workflow_chat.py +++ b/services/workflow_chat/workflow_chat.py @@ -29,6 +29,25 @@ "required": ["yaml", "text"], "additionalProperties": False } + +# Subagent mode (called from global_chat): adds a "handover" field so the model +# can hand a misrouted request back to the caller. It comes FIRST so it is +# generated before yaml/text — streaming can then suppress output and the +# router reroutes before the user sees anything. +_SUBAGENT_OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "handover": { + "anyOf": [ + {"type": "string"}, + {"type": "null"} + ] + }, + **_OUTPUT_SCHEMA["properties"] + }, + "required": ["handover", "yaml", "text"], + "additionalProperties": False +} from anthropic import ( Anthropic, APIConnectionError, @@ -89,6 +108,9 @@ class Payload: stream: Optional[bool] = False read_only: Optional[bool] = False metrics_opt_in: Optional[bool] = None + # Subagent mode: set only when called from global_chat, never by direct + # production callers. Enables the handover response field. + subagent: Optional[bool] = False @classmethod def from_dict(cls, data: Dict[str, Any]) -> "Payload": @@ -107,6 +129,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "Payload": stream=data.get("stream", False), read_only=data.get("read_only", False), metrics_opt_in=data.get("metrics_opt_in"), + subagent=data.get("subagent", False), ) @@ -123,6 +146,8 @@ class ChatResponse: content_yaml: str history: List[Dict[str, str]] usage: Dict[str, Any] + # Subagent mode only: reason the request was handed back to the caller + handover: Optional[str] = None class AnthropicClient: @@ -137,6 +162,9 @@ def __init__(self, config: Optional[ChatConfig] = None): # so restore_components runs once and new-component UUIDs stay identical # between the streamed preview and the persisted payload. self._streamed_yaml = None + # Subagent mode: handover reason parsed from the model's response. + # Set as early as possible while streaming so text output is suppressed. + self._handover = None @staticmethod def _unescape_json_string(text): @@ -160,13 +188,15 @@ def generate( stream: Optional[bool] = False, current_page: Optional[dict] = None, read_only: Optional[bool] = False, + subagent: Optional[bool] = False, + stream_manager: Optional[StreamManager] = None, ) -> ChatResponse: """Generate a response using the Claude API. Retry up to 2 times if YAML/JSON parsing fails.""" - + with sentry_sdk.start_transaction(name="workflow_generation") as transaction: history = history.copy() if history else [] - stream_manager = StreamManager(model=self.config.model, stream=stream) + stream_manager = stream_manager or StreamManager(model=self.config.model, stream=stream) # Extract and preserve existing components (skip in read-only mode) preserved_values = {} @@ -189,14 +219,15 @@ def generate( existing_yaml=processed_existing_yaml, errors=errors, history=history, - read_only=read_only + read_only=read_only, + subagent=subagent ) # Structured outputs config — guarantees valid JSON matching schema output_config = { "format": { "type": "json_schema", - "schema": _OUTPUT_SCHEMA + "schema": _SUBAGENT_OUTPUT_SCHEMA if subagent else _OUTPUT_SCHEMA }, "effort": "medium" } @@ -212,6 +243,7 @@ def generate( for attempt in range(max_retries + 1): # Reset per attempt so a retry never reuses a prior stream's YAML self._streamed_yaml = None + self._handover = None with sentry_sdk.start_span(description="anthropic_api_call"): if stream: logger.info("Making streaming API call") @@ -244,7 +276,7 @@ def generate( message = stream_obj.get_final_message() # Flush any remaining buffered text, stripping JSON closing chars - if text_started: + if text_started and not self._handover: if sent_length < len(accumulated_response): remaining = accumulated_response[sent_length:] remaining = re.sub(r'"\s*}\s*$', '', remaining) @@ -279,6 +311,18 @@ def generate( # If YAML parsing succeeded or we're on the last attempt, return the result if response_yaml is not None or attempt == max_retries: + if self._handover: + logger.info(f"workflow_chat handing over: {self._handover}") + # Deliberately do NOT end the stream: the caller reroutes + # the request and the next agent continues on the same stream. + return ChatResponse( + content=response_text or "", + content_yaml=None, + history=history, + usage=accumulated_usage, + handover=self._handover, + ) + if not response_text: stop_reason = getattr(message, "stop_reason", None) empty_reason = "max_tokens" if stop_reason == "max_tokens" else "no_text_blocks" @@ -427,6 +471,12 @@ def split_format_yaml(self, response, preserved_values=None, stream_manager=None # Try to parse the response as JSON response_data = json.loads(response) + # Subagent mode: a handover means the request is being handed back + # to the caller — capture the reason and skip the YAML entirely + if response_data.get("handover"): + self._handover = response_data["handover"] + return response_data.get("text", "").strip(), "" + # Extract text and yaml from the JSON output_text = response_data.get("text", "").strip() raw_yaml = response_data.get("yaml") or "" @@ -600,15 +650,25 @@ def process_stream_event(self, event, accumulated_response, text_started, sent_l match = re.search(r'"text"\s*:\s*"', accumulated_response) if match: - # Close the partial object and extract the yaml field + # Close the partial object and extract the fields + # generated before "text" (yaml, and in subagent mode + # the handover reason, which comes first) yaml_part = accumulated_response[:match.start()] yaml_raw = yaml_part.rstrip().rstrip(",") + "}" try: - yaml_value = json.loads(yaml_raw).get("yaml") - except (json.JSONDecodeError, ValueError, AttributeError): - yaml_value = None - - if yaml_value: + partial = json.loads(yaml_raw) + except (json.JSONDecodeError, ValueError): + partial = None + if not isinstance(partial, dict): + partial = {} + + if partial.get("handover"): + # Handed back to the caller: suppress all output — + # the rerouted agent produces the user-facing reply + self._handover = partial["handover"] + + yaml_value = partial.get("yaml") + if yaml_value and not self._handover: # Finalize before sending so the streamed preview carries # real IDs/code, not raw placeholders. Cache it so the final # response reuses the identical YAML. Only send if the content @@ -626,7 +686,7 @@ def process_stream_event(self, event, accumulated_response, text_started, sent_l sent_length = match.end() text_started = True - if text_started: + if text_started and not self._handover: # Text phase: stream with buffer for split escape sequences buffer_size = 2 safe_to_send_until = len(accumulated_response) - buffer_size @@ -646,7 +706,7 @@ def main(data_dict: dict) -> dict: """ try: sentry_sdk.set_context("request_data", { - k: v for k, v in data_dict.items() if k != "api_key" + k: v for k, v in data_dict.items() if k not in ("api_key", "_stream_manager") }) data = Payload.from_dict(data_dict) @@ -686,7 +746,11 @@ def main(data_dict: dict) -> dict: history=data.history, stream=data.stream, current_page=current_page, - read_only=data.read_only + read_only=data.read_only, + subagent=data.subagent, + # In-process callers (global_chat) may inject a shared stream + # manager so a handed-over request continues the same stream + stream_manager=data_dict.get("_stream_manager"), ) if tracking: @@ -698,6 +762,12 @@ def main(data_dict: dict) -> dict: if diff_meta: langfuse.update_current_span(metadata=diff_meta) + # Tag the trace when the request was handed back for rerouting to + # the planner, so we can filter for handovers. + if tracking and result.handover: + with propagate_attributes(tags=["handover"]): + pass + # Build response response_dict = { "response": result.content, @@ -707,6 +777,9 @@ def main(data_dict: dict) -> dict: "meta": {"apollo_version": APOLLO_VERSION} } + if result.handover: + response_dict["handover"] = result.handover + return response_dict except ApolloError: diff --git a/services/yaml_utils.py b/services/yaml_utils.py new file mode 100644 index 00000000..17920ec7 --- /dev/null +++ b/services/yaml_utils.py @@ -0,0 +1,202 @@ +""" +Shared utility functions for working with workflow YAML strings. + +Used by global_chat (router, planner, subagent caller) and by job_chat in +subagent mode for job extraction, code stitching, and step inspection. +""" +import re + +import yaml + + +def get_page_view(page: str | None) -> tuple[str | None, str | None]: + """ + Classify what the user has on screen from the `page` breadcrumb — the single + parser for that URL (get_step_name_from_page delegates to this). + + Shapes (names are raw, may contain spaces): + workflows// -> ("step", "") job code page + workflows/ -> ("overview", None) workflow canvas + settings / absent / anything else -> (None, None) + + Because a name may itself contain "/", the returned step name is a + best-effort candidate — the caller must validate it against the workflow + YAML rather than trust it. + """ + if not page: + return None, None + parts = page.strip("/").split("/") + if parts[0] != "workflows": + return None, None + if len(parts) == 2: + return "overview", None + if len(parts) == 3 and parts[2] != "settings": + return "step", parts[2] + return None, None + + +def get_step_name_from_page(page: str | None) -> str | None: + """ + Extract the focused step name from a job-code page URL, or None for the + canvas, settings, or an unrecognized value. + + Examples: + workflows/my-workflow/fetch-patients -> "fetch-patients" + workflows/my-workflow -> None + workflows/my-workflow/settings -> None + """ + view, step = get_page_view(page) + return step if view == "step" else None + + +def normalize_name(name: str) -> str: + """Normalize a name for fuzzy matching: lowercase, non-alphanumeric chars become hyphens.""" + return re.sub(r'[^a-z0-9]', '-', name.lower()).strip('-') + + +def find_job_in_yaml(yaml_str: str, step_name: str) -> tuple[str | None, dict | None]: + """ + Find a job in the workflow YAML by step name. + + Tries direct key match first, then normalized name comparison against + both the job key and the job's name field. + + Returns: + (job_key, job_data) or (None, None) if not found or on parse error + """ + try: + yaml_data = yaml.safe_load(yaml_str) + except Exception: + return None, None + + if not yaml_data or "jobs" not in yaml_data: + return None, None + + jobs = yaml_data["jobs"] + + # Direct key match + if step_name in jobs: + return step_name, jobs[step_name] + + # Normalized match: compare against job key and name field + normalized_step = normalize_name(step_name) + for job_key, job_data in jobs.items(): + if normalize_name(job_key) == normalized_step: + return job_key, job_data + job_name = job_data.get("name", "") + if normalize_name(job_name) == normalized_step: + return job_key, job_data + + return None, None + + +EMPTY_JOB_BODY = "// Add operations here" + + +def workflow_has_job_code(yaml_str: str | None) -> bool: + """Return True if any job has a non-empty, non-placeholder body. + + The canonical empty-job marker is ``// Add operations here`` (see + workflow_chat); a blank body or that marker means "no code yet". Used to + decide whether a "what does this do" question needs the planner (to read the + real code) or can take the faster workflow_agent path (structure only). + """ + try: + yaml_data = yaml.safe_load(yaml_str) + except Exception: + return False + if not yaml_data or "jobs" not in yaml_data: + return False + for job_data in yaml_data["jobs"].values(): + body = (job_data or {}).get("body") + if isinstance(body, str) and body.strip() and body.strip() != EMPTY_JOB_BODY: + return True + return False + + +def redact_job_bodies(yaml_str: str) -> str: + """Return workflow YAML with job bodies replaced by a placeholder and id + fields removed. + + This is the read-only structural view shown to the planner and to job_chat + in subagent mode. It never round-trips back into a real workflow, so the + UUID ids are pure noise to the model — dropping them saves tokens. + """ + try: + yaml_data = yaml.safe_load(yaml_str) + if yaml_data and "jobs" in yaml_data: + _remove_ids(yaml_data) + for job_data in yaml_data["jobs"].values(): + if "body" in job_data: + job_data["body"] = "# [use inspect_job_code to view]" + return yaml.dump(yaml_data, sort_keys=False) + except Exception: + pass + return yaml_str + + +def _remove_ids(obj: object) -> None: + """Recursively remove 'id' keys from a parsed YAML structure.""" + if isinstance(obj, dict): + obj.pop("id", None) + for value in obj.values(): + _remove_ids(value) + elif isinstance(obj, list): + for item in obj: + _remove_ids(item) + + +def stitch_job_code(yaml_str: str, job_key: str, new_code: str) -> str: + """ + Replace a job's body in the workflow YAML with new code. + + Returns the original YAML string unchanged if parsing or stitching fails. + """ + try: + yaml_data = yaml.safe_load(yaml_str) + if yaml_data and "jobs" in yaml_data and job_key in yaml_data["jobs"]: + yaml_data["jobs"][job_key]["body"] = new_code + return yaml.dump(yaml_data, sort_keys=False) + except Exception: + pass + + return yaml_str + + +# Read-only step inspection, shared by the planner and job_chat (subagent +# mode) so both agents explore the workflow with the exact same tool. + +INSPECT_JOB_CODE_TOOL = { + "name": "inspect_job_code", + "description": """Read the current code body of one or more jobs in the workflow (read-only). + +Use this to inspect existing step code before editing — e.g. to find which steps a change applies to before editing only those, or to base one step on another. Pass all the job keys you need in a single call rather than calling once per job.""", + "input_schema": { + "type": "object", + "properties": { + "job_keys": { + "type": "array", + "items": {"type": "string"}, + "description": "The job keys to inspect (e.g. ['fetch-patients', 'load-dhis2'])", + }, + }, + "required": ["job_keys"], + }, +} + + +def inspect_job_code(yaml_str: str | None, job_keys: list[str]) -> str: + """Execute the inspect_job_code tool: return the named jobs' code bodies.""" + if not yaml_str: + return "No workflow available to inspect." + if not job_keys: + return "ERROR: No job keys provided." + + parts = [] + for job_key in job_keys: + _, job_data = find_job_in_yaml(yaml_str, job_key) + if job_data and job_data.get("body"): + parts.append(f"Job code for '{job_key}':\n\n{job_data['body']}") + else: + parts.append(f"No code found for job '{job_key}'.") + return "\n\n".join(parts)